You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by bl...@apache.org on 2013/12/09 03:39:27 UTC
git commit: TAJO-329: Implement physical operator to store in
column-partitioned table. (jaehwa)
Updated Branches:
refs/heads/master 0ddf6ef4a -> 22a6136d3
TAJO-329: Implement physical operator to store in column-partitioned table. (jaehwa)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tajo/commit/22a6136d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tajo/tree/22a6136d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tajo/diff/22a6136d
Branch: refs/heads/master
Commit: 22a6136d33da65f86319aead0a60c34140b11172
Parents: 0ddf6ef
Author: blrunner <jh...@gruter.com>
Authored: Mon Dec 9 11:39:16 2013 +0900
Committer: blrunner <jh...@gruter.com>
Committed: Mon Dec 9 11:39:16 2013 +0900
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../engine/planner/PhysicalPlannerImpl.java | 16 +-
.../ColumnPartitionedTableStoreExec.java | 185 +++++++++++++++++
.../org/apache/tajo/master/GlobalEngine.java | 7 +
.../tajo/engine/query/TestInsertQuery.java | 202 +++++++++++++++++++
5 files changed, 411 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/22a6136d/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 11626ea..19173ed 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -78,6 +78,8 @@ Release 0.8.0 - unreleased
IMPROVEMENTS
+ TAJO-329: Implement physical operator to store in column-partitioned table. (jaehwa)
+
TAJO-391: Change the default type of real values from FLOAT4 to FLOAT8
when parsing the user queries. (DaeMyung Kang via jihoon)
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/22a6136d/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
index 7bc9455..bbecd50 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PhysicalPlannerImpl.java
@@ -73,7 +73,8 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
execPlan = createPlanRecursive(context, logicalPlan);
if (execPlan instanceof StoreTableExec
|| execPlan instanceof IndexedStoreExec
- || execPlan instanceof PartitionedStoreExec) {
+ || execPlan instanceof PartitionedStoreExec
+ || execPlan instanceof ColumnPartitionedTableStoreExec) {
return execPlan;
} else if (context.getDataChannel() != null) {
return buildOutputOperator(context, logicalPlan, execPlan);
@@ -633,6 +634,19 @@ public class PhysicalPlannerImpl implements PhysicalPlanner {
return new TunnelExec(ctx, plan.getOutSchema(), subOp);
}
+ // Find partitioned table
+ if (plan.getPartitions() != null) {
+ if (plan.getPartitions().getPartitionsType().equals(CatalogProtos.PartitionsType.COLUMN)) {
+ return new ColumnPartitionedTableStoreExec(ctx, plan, subOp);
+ } else if (plan.getPartitions().getPartitionsType().equals(CatalogProtos.PartitionsType.HASH)) {
+ // TODO
+ } else if (plan.getPartitions().getPartitionsType().equals(CatalogProtos.PartitionsType.RANGE)) {
+ // TODO
+ } else if (plan.getPartitions().getPartitionsType().equals(CatalogProtos.PartitionsType.LIST)) {
+ // TODO
+ }
+ }
+
return new StoreTableExec(ctx, plan, subOp);
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/22a6136d/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ColumnPartitionedTableStoreExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ColumnPartitionedTableStoreExec.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ColumnPartitionedTableStoreExec.java
new file mode 100644
index 0000000..f7db4bd
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/physical/ColumnPartitionedTableStoreExec.java
@@ -0,0 +1,185 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ *
+ */
+package org.apache.tajo.engine.planner.physical;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.catalog.CatalogUtil;
+import org.apache.tajo.catalog.Column;
+import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.catalog.statistics.StatisticsUtil;
+import org.apache.tajo.catalog.statistics.TableStats;
+import org.apache.tajo.datum.Datum;
+import org.apache.tajo.engine.planner.logical.StoreTableNode;
+import org.apache.tajo.storage.Appender;
+import org.apache.tajo.storage.StorageManagerFactory;
+import org.apache.tajo.storage.StorageUtil;
+import org.apache.tajo.storage.Tuple;
+import org.apache.tajo.worker.TaskAttemptContext;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This class is a physical operator to store at column partitioned table.
+ */
+public class ColumnPartitionedTableStoreExec extends UnaryPhysicalExec {
+ private static Log LOG = LogFactory.getLog(ColumnPartitionedTableStoreExec.class);
+
+ private final TableMeta meta;
+ private final StoreTableNode plan;
+ private Tuple tuple;
+ private Path storeTablePath;
+ private final Map<String, Appender> appenderMap = new HashMap<String, Appender>();
+ private int[] columnIndexes;
+ private String[] columnNames;
+
+
+ /**
+ * @throws java.io.IOException
+ *
+ */
+ public ColumnPartitionedTableStoreExec(TaskAttemptContext context, StoreTableNode plan, PhysicalExec child) throws IOException {
+ super(context, plan.getInSchema(), plan.getOutSchema(), child);
+ this.plan = plan;
+
+ // set table meta
+ if (this.plan.hasOptions()) {
+ meta = CatalogUtil.newTableMeta(plan.getStorageType(), plan.getOptions());
+ } else {
+ meta = CatalogUtil.newTableMeta(plan.getStorageType());
+ }
+
+ // Find column index to name subpartition directory path
+ if (this.plan.getPartitions() != null) {
+ if (this.plan.getPartitions().getColumns() != null) {
+ columnIndexes = new int[plan.getPartitions().getColumns().size()];
+ columnNames = new String[columnIndexes.length];
+ for(int i = 0; i < plan.getPartitions().getColumns().size(); i++) {
+ Column targetColumn = plan.getPartitions().getColumn(i);
+ for(int j = 0; j < plan.getInSchema().getColumns().size();j++) {
+ Column inputColumn = plan.getInSchema().getColumn(j);
+ if (inputColumn.getColumnName().equals(targetColumn.getColumnName())) {
+ columnIndexes[i] = j;
+ columnNames[i] = targetColumn.getColumnName();
+ }
+ }
+ }
+ }
+ }
+ }
+
+ public void init() throws IOException {
+ super.init();
+
+ storeTablePath = context.getOutputPath();
+ FileSystem fs = storeTablePath.getFileSystem(context.getConf());
+ if (!fs.exists(storeTablePath.getParent())) {
+ fs.mkdirs(storeTablePath.getParent());
+ }
+ }
+
+ private Appender getAppender(String partition) throws IOException {
+ Appender appender = appenderMap.get(partition);
+
+ if (appender == null) {
+ Path dataFile = getDataFile(partition);
+ FileSystem fs = dataFile.getFileSystem(context.getConf());
+
+ if (fs.exists(dataFile.getParent())) {
+ LOG.info("Path " + dataFile.getParent() + " already exists!");
+ } else {
+ fs.mkdirs(dataFile.getParent());
+ LOG.info("Add subpartition path directory :" + dataFile.getParent());
+ }
+
+ if (fs.exists(dataFile)) {
+ LOG.info("File " + dataFile + " already exists!");
+ FileStatus status = fs.getFileStatus(dataFile);
+ LOG.info("File size: " + status.getLen());
+ }
+
+ appender = StorageManagerFactory.getStorageManager(context.getConf()).getAppender(meta,
+ outSchema, dataFile);
+ appender.enableStats();
+ appender.init();
+ appenderMap.put(partition, appender);
+ } else {
+ appender = appenderMap.get(partition);
+ }
+ return appender;
+ }
+
+ private Path getDataFile(String partition) {
+ return StorageUtil.concatPath(storeTablePath.getParent(), partition, storeTablePath.getName());
+ }
+
+ /* (non-Javadoc)
+ * @see PhysicalExec#next()
+ */
+ @Override
+ public Tuple next() throws IOException {
+ StringBuilder sb = new StringBuilder();
+ while((tuple = child.next()) != null) {
+ // set subpartition directory name
+ sb.delete(0, sb.length());
+ if (columnIndexes != null) {
+ for(int i = 0; i < columnIndexes.length; i++) {
+ Datum datum = (Datum) tuple.get(columnIndexes[i]);
+ if(i > 0)
+ sb.append("/");
+ sb.append(columnNames[i]).append("=");
+ sb.append(datum.asChars());
+ }
+ }
+
+ // add tuple
+ Appender appender = getAppender(sb.toString());
+ appender.addTuple(tuple);
+ }
+
+ List<TableStats> statSet = new ArrayList<TableStats>();
+ for (Map.Entry<String, Appender> entry : appenderMap.entrySet()) {
+ Appender app = entry.getValue();
+ app.flush();
+ app.close();
+ statSet.add(app.getStats());
+ }
+
+ // Collect and aggregated statistics data
+ TableStats aggregated = StatisticsUtil.aggregateTableStat(statSet);
+ context.setResultStats(aggregated);
+
+ return null;
+ }
+
+ @Override
+ public void rescan() throws IOException {
+ // nothing to do
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/22a6136d/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
index 6773711..6ba93d0 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
@@ -482,6 +482,13 @@ public class GlobalEngine extends AbstractService {
storeNode.setChild(childBlocks.get(0).getRoot());
}
+ // If InsertNode contains table partition information, StoreNode must has it.
+ if (insertNode.hasTargetTable()) {
+ if (insertNode.getTargetTable().getPartitions() != null) {
+ storeNode.setPartitions(insertNode.getTargetTable().getPartitions());
+ }
+ }
+
// find a subquery query of insert node and merge root block and subquery into one query block.
PlannerUtil.replaceNode(plan.getRootBlock().getRoot(), storeNode, NodeType.INSERT);
plan.getRootBlock().refresh();
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/22a6136d/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestInsertQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestInsertQuery.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestInsertQuery.java
index b770de5..b854d31 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestInsertQuery.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestInsertQuery.java
@@ -240,4 +240,206 @@ public class TestInsertQuery {
assertTrue(codec instanceof DeflateCodec);
}
}
+
+ @Test
+ public final void testInsertOverwritePartitionByColumn1() throws Exception {
+ String tableName ="InsertOverwritePartitionByColumn1";
+ ResultSet res = tpch.execute("create table " + tableName +" (col1 int4, col2 int4, col3 float8) partition by column(col1) ");
+ res.close();
+ TajoTestingCluster cluster = tpch.getTestingCluster();
+ CatalogService catalog = cluster.getMaster().getCatalog();
+ assertTrue(catalog.existsTable(tableName));
+
+ res = tpch.execute("insert overwrite into " + tableName
+ + " select l_orderkey, l_partkey, l_quantity from lineitem");
+ res.close();
+
+ TableDesc desc = catalog.getTableDesc(tableName);
+ Path path = desc.getPath();
+
+ FileSystem fs = FileSystem.get(tpch.getTestingCluster().getConfiguration());
+ assertTrue(fs.isDirectory(path));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=1")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=2")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3")));
+ assertEquals(5, desc.getStats().getNumRows().intValue());
+ }
+
+ @Test
+ public final void testInsertOverwritePartitionByColumn2() throws Exception {
+ String tableName ="InsertOverwritePartitionByColumn2";
+ ResultSet res = tpch.execute("create table " + tableName +" (col1 int4, col2 int4, col3 float8) partition by column(col1, col2) ");
+ res.close();
+ TajoTestingCluster cluster = tpch.getTestingCluster();
+ CatalogService catalog = cluster.getMaster().getCatalog();
+ assertTrue(catalog.existsTable(tableName));
+
+ res = tpch.execute("insert overwrite into " + tableName
+ + " select l_orderkey, l_partkey, l_quantity from lineitem");
+ res.close();
+
+ TableDesc desc = catalog.getTableDesc(tableName);
+ Path path = desc.getPath();
+
+ FileSystem fs = FileSystem.get(tpch.getTestingCluster().getConfiguration());
+ assertTrue(fs.isDirectory(path));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=1")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=1/col2=1")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=2")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=2/col2=2")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3/col2=2")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3/col2=3")));
+ assertEquals(5, desc.getStats().getNumRows().intValue());
+ }
+
+ @Test
+ public final void testInsertOverwritePartitionByColumn3() throws Exception {
+ String tableName ="InsertOverwritePartitionByColumn3";
+ ResultSet res = tpch.execute("create table " + tableName +" (col1 int4, col2 int4, col3 float8) partition by column(col1, col2, col3) ");
+ res.close();
+ TajoTestingCluster cluster = tpch.getTestingCluster();
+ CatalogService catalog = cluster.getMaster().getCatalog();
+ assertTrue(catalog.existsTable(tableName));
+
+ res = tpch.execute("insert overwrite into " + tableName
+ + " select l_orderkey, l_partkey, l_quantity from lineitem");
+ res.close();
+
+ TableDesc desc = catalog.getTableDesc(tableName);
+ Path path = desc.getPath();
+
+ FileSystem fs = FileSystem.get(tpch.getTestingCluster().getConfiguration());
+ assertTrue(fs.isDirectory(path));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=1")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=1/col2=1")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=1/col2=1/col3=17.0")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=2")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=2/col2=2")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=2/col2=2/col3=38.0")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3/col2=2")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3/col2=3")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3/col2=2/col3=45.0")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3/col2=3/col3=49.0")));
+ assertEquals(5, desc.getStats().getNumRows().intValue());
+
+ }
+ @Test
+ public final void testInsertOverwritePartitionByColumnWithCompression1() throws Exception {
+ String tableName = "testInsertOverwritePartitionByColumnWithCompression1";
+ ResultSet res = tpch.execute("create table " + tableName + " (col1 int4, col2 int4, col3 float8) USING csv WITH ('csvfile.delimiter'='|','compression.codec'='org.apache.hadoop.io.compress.DeflateCodec') partition by column(col1)");
+ res.close();
+ TajoTestingCluster cluster = tpch.getTestingCluster();
+ CatalogService catalog = cluster.getMaster().getCatalog();
+ assertTrue(catalog.existsTable(tableName));
+
+ res = tpch.execute("insert overwrite into " + tableName + " select l_orderkey, l_partkey, l_quantity from lineitem");
+ res.close();
+ TableDesc desc = catalog.getTableDesc(tableName);
+ assertEquals(5, desc.getStats().getNumRows().intValue());
+
+ FileSystem fs = FileSystem.get(tpch.getTestingCluster().getConfiguration());
+ assertTrue(fs.exists(desc.getPath()));
+ CompressionCodecFactory factory = new CompressionCodecFactory(tpch.getTestingCluster().getConfiguration());
+
+ Path path = desc.getPath();
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=1")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=2")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3")));
+
+ for (FileStatus partition : fs.listStatus(path)){
+ assertTrue(fs.isDirectory(partition.getPath()));
+ for (FileStatus file : fs.listStatus(partition.getPath())) {
+ CompressionCodec codec = factory.getCodec(file.getPath());
+ assertTrue(codec instanceof DeflateCodec);
+ }
+ }
+ }
+
+ @Test
+ public final void testInsertOverwritePartitionByColumnWithCompression2() throws Exception {
+ String tableName = "testInsertOverwritePartitionByColumnWithCompression2";
+ ResultSet res = tpch.execute("create table " + tableName + " (col1 int4, col2 int4, col3 float8) USING csv WITH ('csvfile.delimiter'='|','compression.codec'='org.apache.hadoop.io.compress.DeflateCodec') partition by column(col1, col2)");
+ res.close();
+ TajoTestingCluster cluster = tpch.getTestingCluster();
+ CatalogService catalog = cluster.getMaster().getCatalog();
+ assertTrue(catalog.existsTable(tableName));
+
+ res = tpch.execute("insert overwrite into " + tableName + " select l_orderkey, l_partkey, l_quantity from lineitem");
+ res.close();
+ TableDesc desc = catalog.getTableDesc(tableName);
+ assertEquals(5, desc.getStats().getNumRows().intValue());
+
+ FileSystem fs = FileSystem.get(tpch.getTestingCluster().getConfiguration());
+ assertTrue(fs.exists(desc.getPath()));
+ CompressionCodecFactory factory = new CompressionCodecFactory(tpch.getTestingCluster().getConfiguration());
+
+ Path path = desc.getPath();
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=1")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=1/col2=1")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=2")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=2/col2=2")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3/col2=2")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3/col2=3")));
+
+ for (FileStatus partition1 : fs.listStatus(path)){
+ assertTrue(fs.isDirectory(partition1.getPath()));
+ for (FileStatus partition2 : fs.listStatus(partition1.getPath())) {
+ assertTrue(fs.isDirectory(partition2.getPath()));
+ for (FileStatus file : fs.listStatus(partition2.getPath())) {
+ CompressionCodec codec = factory.getCodec(file.getPath());
+ assertTrue(codec instanceof DeflateCodec);
+ }
+ }
+ }
+ }
+
+ @Test
+ public final void testInsertOverwritePartitionByColumnWithCompression3() throws Exception {
+ String tableName = "testInsertOverwritePartitionByColumnWithCompression3";
+ ResultSet res = tpch.execute("create table " + tableName + " (col1 int4, col2 int4, col3 float8) USING csv WITH ('csvfile.delimiter'='|','compression.codec'='org.apache.hadoop.io.compress.DeflateCodec') partition by column(col1, col2, col3)");
+ res.close();
+ TajoTestingCluster cluster = tpch.getTestingCluster();
+ CatalogService catalog = cluster.getMaster().getCatalog();
+ assertTrue(catalog.existsTable(tableName));
+
+ res = tpch.execute("insert overwrite into " + tableName + " select l_orderkey, l_partkey, l_quantity from lineitem");
+ res.close();
+ TableDesc desc = catalog.getTableDesc(tableName);
+ assertEquals(5, desc.getStats().getNumRows().intValue());
+
+ FileSystem fs = FileSystem.get(tpch.getTestingCluster().getConfiguration());
+ assertTrue(fs.exists(desc.getPath()));
+ CompressionCodecFactory factory = new CompressionCodecFactory(tpch.getTestingCluster().getConfiguration());
+
+ Path path = desc.getPath();
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=1")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=1/col2=1")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=1/col2=1/col3=17.0")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=2")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=2/col2=2")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=2/col2=2/col3=38.0")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3/col2=2")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3/col2=3")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3/col2=2/col3=45.0")));
+ assertTrue(fs.isDirectory(new Path(path.toUri() + "/col1=3/col2=3/col3=49.0")));
+
+ for (FileStatus partition1 : fs.listStatus(path)){
+ assertTrue(fs.isDirectory(partition1.getPath()));
+ for (FileStatus partition2 : fs.listStatus(partition1.getPath())) {
+ assertTrue(fs.isDirectory(partition2.getPath()));
+ for (FileStatus partition3 : fs.listStatus(partition2.getPath())) {
+ assertTrue(fs.isDirectory(partition3.getPath()));
+ for (FileStatus file : fs.listStatus(partition3.getPath())) {
+ CompressionCodec codec = factory.getCodec(file.getPath());
+ assertTrue(codec instanceof DeflateCodec);
+ }
+ }
+ }
+ }
+ }
+
}