TAJO-1658: Filter push down to underlying storages. Closes #613
Project: http://git-wip-us.apache.org/repos/asf/tajo/repo Commit: http://git-wip-us.apache.org/repos/asf/tajo/commit/8f3215d1 Tree: http://git-wip-us.apache.org/repos/asf/tajo/tree/8f3215d1 Diff: http://git-wip-us.apache.org/repos/asf/tajo/diff/8f3215d1 Branch: refs/heads/index_support Commit: 8f3215d12346d71808a43bf4e2a22e1edd8874ff Parents: f57d6c4 Author: Hyunsik Choi <[email protected]> Authored: Sun Jun 28 02:12:11 2015 -0700 Committer: Hyunsik Choi <[email protected]> Committed: Sun Jun 28 02:12:11 2015 -0700 ---------------------------------------------------------------------- CHANGES | 6 +- .../org/apache/tajo/catalog/CatalogUtil.java | 4 +- .../java/org/apache/tajo/catalog/TableMeta.java | 92 +-------------- .../planner/physical/EmptyScanIterator.java | 40 +++++++ .../engine/planner/physical/SeqScanExec.java | 111 ++++++++++-------- .../tajo/engine/utils/TupleCacheScanner.java | 114 ------------------- .../tajo/plan/LogicalPlanPreprocessor.java | 5 +- .../org/apache/tajo/plan/LogicalPlanner.java | 10 +- .../org/apache/tajo/plan/TablePropertyUtil.java | 92 +++++++++++++++ .../org/apache/tajo/plan/util/PlannerUtil.java | 28 ----- .../apache/tajo/storage/AbstractScanner.java | 27 +++-- .../org/apache/tajo/storage/MergeScanner.java | 5 +- .../org/apache/tajo/storage/NullScanner.java | 8 +- .../java/org/apache/tajo/storage/Scanner.java | 9 +- .../apache/tajo/storage/SeekableScanner.java | 4 +- .../apache/tajo/storage/hbase/HBaseScanner.java | 6 +- .../java/org/apache/tajo/storage/CSVFile.java | 4 +- .../org/apache/tajo/storage/FileScanner.java | 6 - .../java/org/apache/tajo/storage/RawFile.java | 7 ++ .../java/org/apache/tajo/storage/RowFile.java | 7 ++ .../apache/tajo/storage/avro/AvroScanner.java | 7 ++ .../tajo/storage/parquet/ParquetScanner.java | 7 ++ .../org/apache/tajo/storage/rcfile/RCFile.java | 7 ++ .../sequencefile/SequenceFileScanner.java | 7 ++ .../tajo/storage/text/DelimitedTextFile.java | 7 +- .../apache/tajo/storage/TestMergeScanner.java | 2 +- .../org/apache/tajo/storage/TestStorages.java | 10 +- 27 files changed, 309 insertions(+), 323 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/CHANGES ---------------------------------------------------------------------- diff --git a/CHANGES b/CHANGES index e0cbd47..04c4392 100644 --- a/CHANGES +++ b/CHANGES @@ -64,8 +64,8 @@ Release 0.11.0 - unreleased TAJO-1603: Refactor StorageManager. (hyunsik) - TAJO-1542: Refactoring of HashJoinExecs. (Contributed by Navis, Committed by - hyunsik) + TAJO-1542: Refactoring of HashJoinExecs. (Contributed by Navis, + Committed by hyunsik) TAJO-1591: Change StoreType represented as Enum to String type. (hyunsik) @@ -357,6 +357,8 @@ Release 0.11.0 - unreleased SUB TASKS + TAJO-1658: Filter push down to underlying storages. (hyunsik) + TAJO-1616: Implement TablespaceManager to load Tablespaces. (hyunsik) TAJO-1615: Implement TaskManager. (jinho) http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java ---------------------------------------------------------------------- diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java index 6c6915b..638ebca 100644 --- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java +++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/CatalogUtil.java @@ -313,7 +313,7 @@ public class CatalogUtil { } public static TableMeta newTableMeta(String storeType) { - KeyValueSet defaultProperties = CatalogUtil.newPhysicalProperties(storeType); + KeyValueSet defaultProperties = CatalogUtil.newDefaultProperty(storeType); return new TableMeta(storeType, defaultProperties); } @@ -871,7 +871,7 @@ public class CatalogUtil { * @param storeType StoreType * @return Table properties */ - public static KeyValueSet newPhysicalProperties(String storeType) { + public static KeyValueSet newDefaultProperty(String storeType) { KeyValueSet options = new KeyValueSet(); if (storeType.equalsIgnoreCase("CSV") || storeType.equalsIgnoreCase("TEXT")) { options.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER); http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/TableMeta.java ---------------------------------------------------------------------- diff --git a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/TableMeta.java b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/TableMeta.java index 2b31b83..6838fe0 100644 --- a/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/TableMeta.java +++ b/tajo-catalog/tajo-catalog-common/src/main/java/org/apache/tajo/catalog/TableMeta.java @@ -24,9 +24,7 @@ import com.google.gson.GsonBuilder; import com.google.gson.annotations.Expose; import org.apache.tajo.catalog.json.CatalogGsonHelper; import org.apache.tajo.catalog.proto.CatalogProtos; -import org.apache.tajo.catalog.proto.CatalogProtos.StoreType; import org.apache.tajo.catalog.proto.CatalogProtos.TableProto; -import org.apache.tajo.catalog.proto.CatalogProtos.TableProtoOrBuilder; import org.apache.tajo.common.ProtoObject; import org.apache.tajo.json.GsonObject; import org.apache.tajo.util.KeyValueSet; @@ -37,95 +35,44 @@ import java.util.Map; * It contains all information for scanning a fragmented table */ public class TableMeta implements ProtoObject<CatalogProtos.TableProto>, GsonObject, Cloneable { - protected TableProto.Builder builder = null; - private TableProto proto = TableProto.getDefaultInstance(); - private boolean viaProto = false; - @Expose protected String storeType; @Expose protected KeyValueSet options; - private TableMeta() { - builder = TableProto.newBuilder(); - } - public TableMeta(String storeType, KeyValueSet options) { - this(); this.storeType = storeType; this.options = new KeyValueSet(options); } public TableMeta(TableProto proto) { - this.proto = proto; - viaProto = true; + this.storeType = proto.getStoreType(); + this.options = new KeyValueSet(proto.getParams()); } public String getStoreType() { - TableProtoOrBuilder p = viaProto ? proto : builder; - if (this.storeType != null) { - return storeType; - } - if (!p.hasStoreType()) { - return null; - } - this.storeType = p.getStoreType(); return this.storeType; } public void setOptions(KeyValueSet options) { - maybeInitBuilder(); this.options = options; } public void putOption(String key, String val) { - maybeInitBuilder(); options.set(key, val); } public boolean containsOption(String key) { - TableProtoOrBuilder p = viaProto ? proto : builder; - if (options != null) { - return this.options.containsKey(key); - } - if (!p.hasParams()) { - return false; - } - this.options = new KeyValueSet(p.getParams()); return options.containsKey(key); } public String getOption(String key) { - TableProtoOrBuilder p = viaProto ? proto : builder; - if (options != null) { - return this.options.get(key); - } - if (!p.hasParams()) { - return null; - } - this.options = new KeyValueSet(p.getParams()); return options.get(key); } public String getOption(String key, String defaultValue) { - TableProtoOrBuilder p = viaProto ? proto : builder; - if (options != null) { - return this.options.get(key, defaultValue); - } - if (!p.hasParams()) { - return null; - } - this.options = new KeyValueSet(p.getParams()); return options.get(key, defaultValue); } public KeyValueSet getOptions() { - TableProtoOrBuilder p = viaProto ? proto : builder; - if (options != null) { - return this.options; - } - if (!p.hasParams()) { - return null; - } - this.options = new KeyValueSet(p.getParams()); return options; } @@ -152,10 +99,8 @@ public class TableMeta implements ProtoObject<CatalogProtos.TableProto>, GsonObj @Override public Object clone() throws CloneNotSupportedException { TableMeta meta = (TableMeta) super.clone(); - meta.builder = TableProto.newBuilder(); meta.storeType = getStoreType(); meta.options = (KeyValueSet) (toMap() != null ? options.clone() : null); - return meta; } @@ -169,10 +114,10 @@ public class TableMeta implements ProtoObject<CatalogProtos.TableProto>, GsonObj // ProtoObject //////////////////////////////////////////////////////////////////////// public TableProto getProto() { - mergeLocalToProto(); - proto = viaProto ? proto : builder.build(); - viaProto = true; - return proto; + TableProto.Builder builder = TableProto.newBuilder(); + builder.setStoreType(storeType); + builder.setParams(options.getProto()); + return builder.build(); } @Override @@ -185,29 +130,4 @@ public class TableMeta implements ProtoObject<CatalogProtos.TableProto>, GsonObj getStoreType(); toMap(); } - - private void maybeInitBuilder() { - if (viaProto || builder == null) { - builder = TableProto.newBuilder(proto); - } - viaProto = true; - } - - private void mergeLocalToBuilder() { - if (storeType != null) { - builder.setStoreType(storeType); - } - if (this.options != null) { - builder.setParams(options.getProto()); - } - } - - private void mergeLocalToProto() { - if(viaProto) { - maybeInitBuilder(); - } - mergeLocalToBuilder(); - proto = builder.build(); - viaProto = true; - } } http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/EmptyScanIterator.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/EmptyScanIterator.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/EmptyScanIterator.java new file mode 100644 index 0000000..cfbd3ed --- /dev/null +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/EmptyScanIterator.java @@ -0,0 +1,40 @@ +/* + * 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.tajo.storage.Tuple; + +import java.io.IOException; + +/** + * Empty Iterator + */ +public class EmptyScanIterator implements ScanIterator { + + @Override + public boolean hasNext() throws IOException { + return false; + } + + @Override + public Tuple next() throws IOException { + throw new IOException(this.getClass().getSimpleName() + "::next() is invoked."); + } +} + http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java index 79e0a5d..b4f7a38 100644 --- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java +++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/SeqScanExec.java @@ -19,10 +19,7 @@ package org.apache.tajo.engine.planner.physical; import org.apache.hadoop.io.IOUtils; -import org.apache.tajo.catalog.Column; -import org.apache.tajo.catalog.Schema; -import org.apache.tajo.catalog.SchemaUtil; -import org.apache.tajo.catalog.TableMeta; +import org.apache.tajo.catalog.*; import org.apache.tajo.catalog.partition.PartitionMethodDesc; import org.apache.tajo.catalog.proto.CatalogProtos; import org.apache.tajo.catalog.statistics.TableStats; @@ -37,7 +34,6 @@ import org.apache.tajo.plan.expr.EvalTreeUtil; import org.apache.tajo.plan.expr.FieldEval; import org.apache.tajo.plan.logical.ScanNode; import org.apache.tajo.plan.rewrite.rules.PartitionedTableRewriter; -import org.apache.tajo.plan.util.PlannerUtil; import org.apache.tajo.storage.*; import org.apache.tajo.storage.fragment.FileFragment; import org.apache.tajo.storage.fragment.FragmentConvertor; @@ -137,8 +133,7 @@ public class SeqScanExec extends ScanExec { } } - @Override - public void init() throws IOException { + public Schema getProjectSchema() { Schema projected; // in the case where projected column or expression are given @@ -167,23 +162,55 @@ public class SeqScanExec extends ScanExec { projected = outSchema; } - initScanner(projected); - super.init(); - - if (plan.hasQual()) { - if (scanner.isProjectable()) { - qual.bind(context.getEvalContext(), projected); - } else { - qual.bind(context.getEvalContext(), inSchema); - } + return projected; + } + private void initScanIterator() { + // We should use FilterScanIterator only if underlying storage does not support filter push down. + if (plan.hasQual() && !scanner.isSelectable()) { scanIt = new FilterScanIterator(scanner, qual); + } else { + if (scanner.isSelectable()) { // TODO - isSelectable should be moved to FormatProperty + scanner.setFilter(qual); + } scanIt = new FullScanIterator(scanner); } } @Override + public void init() throws IOException { + + // Why we should check nullity? See https://issues.apache.org/jira/browse/TAJO-1422 + + if (fragments == null) { + scanIt = new EmptyScanIterator(); + + } else { + Schema projectedFields = getProjectSchema(); + initScanner(projectedFields); + + // See Scanner.isProjectable() method. Depending on the result of isProjectable(), + // the width of retrieved tuple is changed. + // + // If projectable, the retrieved tuple will contain only projected fields. + // Otherwise, the retrieved tuple will contain projected fields and NullDatum + // for non-projected fields. + Schema actualInSchema = scanner.isProjectable() ? projectedFields : inSchema; + + this.projector = new Projector(context, actualInSchema, outSchema, plan.getTargets()); + + if (plan.hasQual()) { + qual.bind(context.getEvalContext(), actualInSchema); + } + + initScanIterator(); + } + + super.init(); + } + + @Override protected void compile() throws CompilationError { if (plan.hasQual()) { qual = context.getPrecompiledEval(inSchema, qual); @@ -191,47 +218,33 @@ public class SeqScanExec extends ScanExec { } private void initScanner(Schema projected) throws IOException { - - TableMeta meta; - try { - meta = (TableMeta) plan.getTableDesc().getMeta().clone(); - } catch (CloneNotSupportedException e) { - throw new RuntimeException(e); - } + TableDesc table = plan.getTableDesc(); + TableMeta meta = table.getMeta(); - // set system default properties - PlannerUtil.applySystemDefaultToTableProperties(context.getQueryContext(), meta); + if (fragments.length > 1) { - // Why we should check nullity? See https://issues.apache.org/jira/browse/TAJO-1422 - if (fragments != null) { - if (fragments.length > 1) { - this.scanner = new MergeScanner(context.getConf(), plan.getPhysicalSchema(), meta, - FragmentConvertor.convert(context.getConf(), fragments), projected - ); - } else { - Tablespace tablespace = TablespaceManager.get(plan.getTableDesc().getUri()).get(); - this.scanner = tablespace.getScanner(meta, plan.getPhysicalSchema(), fragments[0], projected); - } - scanner.init(); + this.scanner = new MergeScanner( + context.getConf(), + plan.getPhysicalSchema(), meta, + FragmentConvertor.convert(context.getConf(), fragments), + projected + ); - // See Scanner.isProjectable() method Depending on the result of isProjectable(), - // the width of retrieved tuple is changed. - // - // If TRUE, the retrieved tuple will contain only projected fields. - // If FALSE, the retrieved tuple will contain projected fields and NullDatum for non-projected fields. - if (scanner.isProjectable()) { - this.projector = new Projector(context, projected, outSchema, plan.getTargets()); - } else { - this.projector = new Projector(context, inSchema, outSchema, plan.getTargets()); - } + } else { + + Tablespace tablespace = TablespaceManager.get(table.getUri()).get(); + this.scanner = tablespace.getScanner( + meta, + plan.getPhysicalSchema(), + fragments[0], + projected); } + + scanner.init(); } @Override public Tuple next() throws IOException { - if (fragments == null) { - return null; - } while(scanIt.hasNext()) { Tuple outTuple = new VTuple(outColumnNum); http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleCacheScanner.java ---------------------------------------------------------------------- diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleCacheScanner.java b/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleCacheScanner.java deleted file mode 100644 index 0fd2fbe..0000000 --- a/tajo-core/src/main/java/org/apache/tajo/engine/utils/TupleCacheScanner.java +++ /dev/null @@ -1,114 +0,0 @@ -/** - * 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.utils; - -import org.apache.tajo.catalog.Column; -import org.apache.tajo.catalog.Schema; -import org.apache.tajo.catalog.statistics.TableStats; -import org.apache.tajo.storage.Scanner; -import org.apache.tajo.storage.Tuple; - -import java.io.IOException; -import java.util.Iterator; -import java.util.List; - -public class TupleCacheScanner implements Scanner { - List<Tuple> cacheData; - Schema schema; - Iterator<Tuple> it; - int count; - TableStats inputStats = new TableStats(); - - public TupleCacheScanner(List<Tuple> cacheData, Schema schema) { - this.cacheData = cacheData; - this.schema = schema; - } - @Override - public void init() throws IOException { - inputStats.setNumRows(cacheData.size()); - inputStats.setReadBytes(0); - it = cacheData.iterator(); - count = 0; - } - - @Override - public Tuple next() throws IOException { - if (it.hasNext()) { - count++; - Tuple tuple = it.next(); - try { - return (Tuple)tuple.clone(); - } catch (CloneNotSupportedException e) { - throw new IOException(e.getMessage(), e); - } - } else { - return null; - } - } - - @Override - public void reset() throws IOException { - init(); - } - - @Override - public void close() throws IOException { - } - - @Override - public boolean isProjectable() { - return false; - } - - @Override - public void setTarget(Column[] targets) { - } - - @Override - public boolean isSelectable() { - return true; - } - - @Override - public void setSearchCondition(Object expr) { - } - - @Override - public boolean isSplittable() { - return false; - } - - @Override - public float getProgress() { - if (cacheData.size() == 0) { - return 1.0f; - } - return ((float)count) / cacheData.size(); - } - - @Override - public TableStats getInputStats() { - return inputStats; - } - - @Override - public Schema getSchema() { - return schema; - } -} http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanPreprocessor.java ---------------------------------------------------------------------- diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanPreprocessor.java b/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanPreprocessor.java index fbad76e..dced4d3 100644 --- a/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanPreprocessor.java +++ b/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanPreprocessor.java @@ -383,14 +383,17 @@ public class LogicalPlanPreprocessor extends BaseAlgebraVisitor<LogicalPlanner.P } TableDesc desc = catalog.getTableDesc(actualRelationName); + ScanNode scanNode = ctx.plan.createNode(ScanNode.class); if (relation.hasAlias()) { scanNode.init(desc, relation.getAlias()); } else { scanNode.init(desc); } - ctx.queryBlock.addRelation(scanNode); + TablePropertyUtil.setTableProperty(ctx.getQueryContext(), scanNode); + + ctx.queryBlock.addRelation(scanNode); return scanNode; } http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java ---------------------------------------------------------------------- diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java b/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java index a2480c9..c51d068 100644 --- a/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java +++ b/tajo-plan/src/main/java/org/apache/tajo/plan/LogicalPlanner.java @@ -1792,20 +1792,20 @@ public class LogicalPlanner extends BaseAlgebraVisitor<LogicalPlanner.PlanContex } // Set default storage properties to table - KeyValueSet properties = CatalogUtil.newPhysicalProperties(createTableNode.getStorageType()); + createTableNode.setOptions(CatalogUtil.newDefaultProperty(createTableNode.getStorageType())); // Priority to apply table properties // 1. Explicit table properties specified in WITH clause // 2. Session variables // Set session variables to properties - PlannerUtil.applySessionToTableProperties(context.queryContext, createTableNode.getStorageType(), properties); - // Set table properties specified in WITH clause + TablePropertyUtil.setTableProperty(context.queryContext, createTableNode); + + // Set table property specified in WITH clause and it will override all others if (expr.hasParams()) { - properties.putAll(expr.getParams()); + createTableNode.getOptions().putAll(expr.getParams()); } - createTableNode.setOptions(properties); if (expr.hasPartition()) { if (expr.getPartitionMethod().getPartitionType().equals(PartitionType.COLUMN)) { http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-plan/src/main/java/org/apache/tajo/plan/TablePropertyUtil.java ---------------------------------------------------------------------- diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/TablePropertyUtil.java b/tajo-plan/src/main/java/org/apache/tajo/plan/TablePropertyUtil.java new file mode 100644 index 0000000..5576889 --- /dev/null +++ b/tajo-plan/src/main/java/org/apache/tajo/plan/TablePropertyUtil.java @@ -0,0 +1,92 @@ +/* + * 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.plan; + +import org.apache.tajo.OverridableConf; +import org.apache.tajo.SessionVars; +import org.apache.tajo.catalog.TableMeta; +import org.apache.tajo.conf.TajoConf; +import org.apache.tajo.plan.logical.CreateTableNode; +import org.apache.tajo.plan.logical.ScanNode; +import org.apache.tajo.storage.StorageConstants; +import org.apache.tajo.util.KeyValueSet; + +/** + * An utility class for table property + */ +public class TablePropertyUtil { + /** + * It sets default table property for newly created table + * + * @param context QueryContext + * @param node CreateTableNode + */ + public static void setTableProperty(OverridableConf context, CreateTableNode node) { + String storeType = node.getStorageType(); + KeyValueSet property = node.getOptions(); + + if (storeType.equalsIgnoreCase("CSV") || storeType.equalsIgnoreCase("TEXT")) { + setSessionToProperty(context, SessionVars.NULL_CHAR, property, StorageConstants.TEXT_NULL); + } + + setSessionToProperty(context, SessionVars.TIMEZONE, property, StorageConstants.TIMEZONE); + } + + private static void setSessionToProperty(OverridableConf context, + SessionVars sessionVarKey, + KeyValueSet property, + String propertyKey) { + + if (context.containsKey(sessionVarKey)) { + property.set(propertyKey, context.get(sessionVarKey)); + } + } + + /** + * It sets default table properties affected by system global configuration + * The table property are implicitly used to read Table rows. + * + * @param context QueryContext + * @param node ScanNode + */ + public static void setTableProperty(OverridableConf context, ScanNode node) { + TableMeta meta = node.getTableDesc().getMeta(); + + setProperty(context, SessionVars.TIMEZONE, meta, StorageConstants.TIMEZONE); + setProperty(context, SessionVars.NULL_CHAR, meta, StorageConstants.TEXT_NULL); + } + + /** + * If there is no table property for the propertyKey, set default property to the table. + * If session variable is set, it is set to the table property. Otherwise, the default property + * in the system conf will be used. + * + * @param context QueryContext + * @param sessionVarKey session variable key + * @param meta TableMeta + * @param propertyKey table property key + */ + private static void setProperty(OverridableConf context, SessionVars sessionVarKey, + TableMeta meta, String propertyKey) { + + if (!meta.containsOption(propertyKey)) { + meta.putOption(propertyKey, context.get(sessionVarKey)); + } + } +} http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-plan/src/main/java/org/apache/tajo/plan/util/PlannerUtil.java ---------------------------------------------------------------------- diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/util/PlannerUtil.java b/tajo-plan/src/main/java/org/apache/tajo/plan/util/PlannerUtil.java index 441e047..19e6ad1 100644 --- a/tajo-plan/src/main/java/org/apache/tajo/plan/util/PlannerUtil.java +++ b/tajo-plan/src/main/java/org/apache/tajo/plan/util/PlannerUtil.java @@ -850,34 +850,6 @@ public class PlannerUtil { return explains.toString(); } - public static void applySessionToTableProperties(OverridableConf sessionVars, - String storeType, - KeyValueSet tableProperties) { - if (storeType.equalsIgnoreCase("CSV") || storeType.equalsIgnoreCase("TEXT")) { - if (sessionVars.containsKey(SessionVars.NULL_CHAR)) { - tableProperties.set(StorageConstants.TEXT_NULL, sessionVars.get(SessionVars.NULL_CHAR)); - } - - if (sessionVars.containsKey(SessionVars.TIMEZONE)) { - tableProperties.set(StorageConstants.TIMEZONE, sessionVars.get(SessionVars.TIMEZONE)); - } - } - } - - /** - * This method sets a set of table properties by System default configs. - * These properties are implicitly used to read or write rows in Table. - * Don't use this method for TableMeta to be stored in Catalog. - * - * @param systemConf System configuration - * @param meta TableMeta to be set - */ - public static void applySystemDefaultToTableProperties(OverridableConf systemConf, TableMeta meta) { - if (!meta.containsOption(StorageConstants.TIMEZONE)) { - meta.putOption(StorageConstants.TIMEZONE, systemConf.get(SessionVars.TIMEZONE)); - } - } - public static boolean isFileStorageType(String storageType) { if (storageType.equalsIgnoreCase("hbase")) { return false; http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/AbstractScanner.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/AbstractScanner.java b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/AbstractScanner.java index 3719412..07fe353 100644 --- a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/AbstractScanner.java +++ b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/AbstractScanner.java @@ -21,60 +21,69 @@ package org.apache.tajo.storage; import org.apache.tajo.catalog.Column; import org.apache.tajo.catalog.Schema; import org.apache.tajo.catalog.statistics.TableStats; +import org.apache.tajo.exception.UnimplementedException; +import org.apache.tajo.exception.UnsupportedException; +import org.apache.tajo.plan.expr.EvalNode; import java.io.IOException; -// dummy scanner +/** + * It's a dummy class to avoid subclass to implement all methods. + */ public abstract class AbstractScanner implements Scanner { @Override public void init() throws IOException { - + throw new UnimplementedException(); } @Override public void reset() throws IOException { + throw new UnimplementedException(); } @Override public void close() throws IOException { + throw new UnimplementedException(); } @Override public boolean isProjectable() { - return false; + throw new UnimplementedException(); } @Override public void setTarget(Column[] targets) { + throw new UnsupportedException(); } @Override public boolean isSelectable() { - return false; + throw new UnimplementedException(); } @Override - public void setSearchCondition(Object expr) { + public void setFilter(EvalNode filter) { + throw new UnsupportedException(); } @Override public boolean isSplittable() { - return false; + throw new UnimplementedException(); } @Override public float getProgress() { - return 0; + throw new UnimplementedException(); } @Override public TableStats getInputStats() { - return null; + throw new UnimplementedException(); } @Override public Schema getSchema() { - return null; + throw new UnimplementedException(); } } http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/MergeScanner.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/MergeScanner.java b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/MergeScanner.java index 67a2f86..87f0310 100644 --- a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/MergeScanner.java +++ b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/MergeScanner.java @@ -25,6 +25,8 @@ import org.apache.tajo.catalog.TableMeta; import org.apache.tajo.catalog.statistics.ColumnStats; import org.apache.tajo.catalog.statistics.TableStats; import org.apache.tajo.conf.TajoConf; +import org.apache.tajo.exception.UnsupportedException; +import org.apache.tajo.plan.expr.EvalNode; import org.apache.tajo.storage.fragment.Fragment; import java.io.IOException; @@ -165,7 +167,8 @@ public class MergeScanner implements Scanner { } @Override - public void setSearchCondition(Object expr) { + public void setFilter(EvalNode filter) { + throw new UnsupportedException(); } @Override http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/NullScanner.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/NullScanner.java b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/NullScanner.java index 4272228..83d8e24 100644 --- a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/NullScanner.java +++ b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/NullScanner.java @@ -21,6 +21,8 @@ import org.apache.tajo.catalog.Column; import org.apache.tajo.catalog.Schema; import org.apache.tajo.catalog.TableMeta; import org.apache.tajo.catalog.statistics.TableStats; +import org.apache.tajo.exception.UnsupportedException; +import org.apache.tajo.plan.expr.EvalNode; import org.apache.tajo.storage.fragment.Fragment; import java.io.IOException; @@ -79,12 +81,12 @@ public class NullScanner implements Scanner { @Override public boolean isSelectable() { - return true; + return false; } @Override - public void setSearchCondition(Object expr) { - + public void setFilter(EvalNode filter) { + throw new UnsupportedException(); } @Override http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/Scanner.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/Scanner.java b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/Scanner.java index 7af8247..2fcb2fd 100644 --- a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/Scanner.java +++ b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/Scanner.java @@ -21,6 +21,7 @@ package org.apache.tajo.storage; import org.apache.tajo.catalog.Column; import org.apache.tajo.catalog.SchemaObject; import org.apache.tajo.catalog.statistics.TableStats; +import org.apache.tajo.plan.expr.EvalNode; import java.io.Closeable; import java.io.IOException; @@ -28,8 +29,8 @@ import java.io.IOException; /** * Scanner Interface */ -public interface Scanner extends SchemaObject, Closeable { +public interface Scanner extends SchemaObject, Closeable { void init() throws IOException; /** @@ -88,12 +89,12 @@ public interface Scanner extends SchemaObject, Closeable { boolean isSelectable(); /** - * Set a search condition - * @param expr to be searched + * Set a filter condition + * @param filter to be searched * * TODO - to be changed Object type */ - void setSearchCondition(Object expr); + void setFilter(EvalNode filter); /** * It returns if the file is splittable. http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/SeekableScanner.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/SeekableScanner.java b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/SeekableScanner.java index 894e7ee..41d0872 100644 --- a/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/SeekableScanner.java +++ b/tajo-storage/tajo-storage-common/src/main/java/org/apache/tajo/storage/SeekableScanner.java @@ -22,7 +22,7 @@ import java.io.IOException; public interface SeekableScanner extends Scanner { - public abstract long getNextOffset() throws IOException; + long getNextOffset() throws IOException; - public abstract void seek(long offset) throws IOException; + void seek(long offset) throws IOException; } http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-hbase/src/main/java/org/apache/tajo/storage/hbase/HBaseScanner.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-hbase/src/main/java/org/apache/tajo/storage/hbase/HBaseScanner.java b/tajo-storage/tajo-storage-hbase/src/main/java/org/apache/tajo/storage/hbase/HBaseScanner.java index 7369897..11851ec 100644 --- a/tajo-storage/tajo-storage-hbase/src/main/java/org/apache/tajo/storage/hbase/HBaseScanner.java +++ b/tajo-storage/tajo-storage-hbase/src/main/java/org/apache/tajo/storage/hbase/HBaseScanner.java @@ -36,6 +36,8 @@ import org.apache.tajo.conf.TajoConf; import org.apache.tajo.datum.Datum; import org.apache.tajo.datum.NullDatum; import org.apache.tajo.datum.TextDatum; +import org.apache.tajo.exception.UnsupportedException; +import org.apache.tajo.plan.expr.EvalNode; import org.apache.tajo.storage.*; import org.apache.tajo.storage.fragment.Fragment; import org.apache.tajo.util.BytesUtils; @@ -424,8 +426,8 @@ public class HBaseScanner implements Scanner { } @Override - public void setSearchCondition(Object expr) { - // TODO implements adding column filter to scanner. + public void setFilter(EvalNode filter) { + throw new UnsupportedException(); } @Override http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/CSVFile.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/CSVFile.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/CSVFile.java index ee3095c..8b8ca76 100644 --- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/CSVFile.java +++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/CSVFile.java @@ -35,6 +35,7 @@ import org.apache.tajo.catalog.statistics.TableStats; import org.apache.tajo.conf.TajoConf; import org.apache.tajo.datum.NullDatum; import org.apache.tajo.exception.UnsupportedException; +import org.apache.tajo.plan.expr.EvalNode; import org.apache.tajo.storage.compress.CodecPool; import org.apache.tajo.storage.exception.AlreadyExistsStorageException; import org.apache.tajo.storage.fragment.Fragment; @@ -536,7 +537,8 @@ public class CSVFile { } @Override - public void setSearchCondition(Object expr) { + public void setFilter(EvalNode filter) { + throw new UnsupportedException(); } @Override http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/FileScanner.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/FileScanner.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/FileScanner.java index 0726125..8844fa5 100644 --- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/FileScanner.java +++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/FileScanner.java @@ -89,12 +89,6 @@ public abstract class FileScanner implements Scanner { this.targets = targets; } - public void setSearchCondition(Object expr) { - if (inited) { - throw new IllegalStateException("Should be called before init()"); - } - } - public static FileSystem getFileSystem(TajoConf tajoConf, Path path) throws IOException { String tajoUser = tajoConf.getVar(TajoConf.ConfVars.USERNAME); FileSystem fs; http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RawFile.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RawFile.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RawFile.java index 4e9bcda..3b655be 100644 --- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RawFile.java +++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RawFile.java @@ -33,6 +33,8 @@ import org.apache.tajo.common.TajoDataTypes.DataType; import org.apache.tajo.datum.DatumFactory; import org.apache.tajo.datum.NullDatum; import org.apache.tajo.datum.ProtobufDatumFactory; +import org.apache.tajo.exception.UnsupportedException; +import org.apache.tajo.plan.expr.EvalNode; import org.apache.tajo.storage.fragment.Fragment; import org.apache.tajo.unit.StorageUnit; import org.apache.tajo.util.BitArray; @@ -426,6 +428,11 @@ public class RawFile { } @Override + public void setFilter(EvalNode filter) { + throw new UnsupportedException(); + } + + @Override public boolean isSplittable(){ return false; } http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RowFile.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RowFile.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RowFile.java index 0e628d4..2be2ec0 100644 --- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RowFile.java +++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RowFile.java @@ -34,6 +34,8 @@ import org.apache.tajo.catalog.statistics.TableStats; import org.apache.tajo.conf.TajoConf.ConfVars; import org.apache.tajo.datum.Datum; import org.apache.tajo.datum.DatumFactory; +import org.apache.tajo.exception.UnsupportedException; +import org.apache.tajo.plan.expr.EvalNode; import org.apache.tajo.storage.exception.AlreadyExistsStorageException; import org.apache.tajo.storage.fragment.Fragment; import org.apache.tajo.util.BitArray; @@ -299,6 +301,11 @@ public class RowFile { } @Override + public void setFilter(EvalNode filter) { + throw new UnsupportedException(); + } + + @Override public boolean isSplittable(){ return true; } http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/avro/AvroScanner.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/avro/AvroScanner.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/avro/AvroScanner.java index 26083a5..729c237 100644 --- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/avro/AvroScanner.java +++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/avro/AvroScanner.java @@ -35,6 +35,8 @@ import org.apache.tajo.catalog.TableMeta; import org.apache.tajo.common.TajoDataTypes; import org.apache.tajo.common.TajoDataTypes.DataType; import org.apache.tajo.datum.*; +import org.apache.tajo.exception.UnsupportedException; +import org.apache.tajo.plan.expr.EvalNode; import org.apache.tajo.storage.FileScanner; import org.apache.tajo.storage.Tuple; import org.apache.tajo.storage.VTuple; @@ -273,6 +275,11 @@ public class AvroScanner extends FileScanner { return false; } + @Override + public void setFilter(EvalNode filter) { + throw new UnsupportedException(); + } + /** * Returns whether this scanner is splittable. * http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/parquet/ParquetScanner.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/parquet/ParquetScanner.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/parquet/ParquetScanner.java index 2f8efcf..822151a 100644 --- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/parquet/ParquetScanner.java +++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/parquet/ParquetScanner.java @@ -21,6 +21,8 @@ package org.apache.tajo.storage.parquet; import org.apache.hadoop.conf.Configuration; import org.apache.tajo.catalog.Schema; import org.apache.tajo.catalog.TableMeta; +import org.apache.tajo.exception.UnsupportedException; +import org.apache.tajo.plan.expr.EvalNode; import org.apache.tajo.storage.FileScanner; import org.apache.tajo.storage.Tuple; import org.apache.tajo.storage.fragment.Fragment; @@ -107,6 +109,11 @@ public class ParquetScanner extends FileScanner { return false; } + @Override + public void setFilter(EvalNode filter) { + throw new UnsupportedException(); + } + /** * Returns whether this scanner is splittable. * http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/rcfile/RCFile.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/rcfile/RCFile.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/rcfile/RCFile.java index 1dcec5f..286ee3a 100644 --- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/rcfile/RCFile.java +++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/rcfile/RCFile.java @@ -37,6 +37,8 @@ import org.apache.tajo.catalog.statistics.TableStats; import org.apache.tajo.conf.TajoConf; import org.apache.tajo.datum.Datum; import org.apache.tajo.datum.NullDatum; +import org.apache.tajo.exception.UnsupportedException; +import org.apache.tajo.plan.expr.EvalNode; import org.apache.tajo.storage.*; import org.apache.tajo.storage.fragment.Fragment; @@ -1787,6 +1789,11 @@ public class RCFile { } @Override + public void setFilter(EvalNode filter) { + throw new UnsupportedException(); + } + + @Override public boolean isSplittable() { return true; } http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileScanner.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileScanner.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileScanner.java index ff73a1c..340e2fa 100644 --- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileScanner.java +++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileScanner.java @@ -31,6 +31,8 @@ import org.apache.tajo.catalog.TableMeta; import org.apache.tajo.conf.TajoConf; import org.apache.tajo.datum.Datum; import org.apache.tajo.datum.NullDatum; +import org.apache.tajo.exception.UnsupportedException; +import org.apache.tajo.plan.expr.EvalNode; import org.apache.tajo.storage.*; import org.apache.tajo.storage.fragment.Fragment; import org.apache.tajo.util.BytesUtils; @@ -332,6 +334,11 @@ public class SequenceFileScanner extends FileScanner { } @Override + public void setFilter(EvalNode filter) { + throw new UnsupportedException(); + } + + @Override public boolean isSplittable(){ return true; } http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/text/DelimitedTextFile.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/text/DelimitedTextFile.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/text/DelimitedTextFile.java index 55a2b96..2aa6707 100644 --- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/text/DelimitedTextFile.java +++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/text/DelimitedTextFile.java @@ -34,6 +34,8 @@ import org.apache.tajo.TaskAttemptId; import org.apache.tajo.catalog.Schema; import org.apache.tajo.catalog.TableMeta; import org.apache.tajo.catalog.statistics.TableStats; +import org.apache.tajo.exception.UnsupportedException; +import org.apache.tajo.plan.expr.EvalNode; import org.apache.tajo.storage.*; import org.apache.tajo.storage.compress.CodecPool; import org.apache.tajo.storage.exception.AlreadyExistsStorageException; @@ -89,7 +91,7 @@ public class DelimitedTextFile { serdeClass = (Class<? extends TextLineSerDe>) Class.forName(serDeClassName); serdeClassCache.put(serDeClassName, serdeClass); } - lineSerder = (TextLineSerDe) ReflectionUtil.newInstance(serdeClass); + lineSerder = ReflectionUtil.newInstance(serdeClass); } catch (Throwable e) { throw new RuntimeException("TextLineSerde class cannot be initialized.", e); } @@ -459,7 +461,8 @@ public class DelimitedTextFile { } @Override - public void setSearchCondition(Object expr) { + public void setFilter(EvalNode filter) { + throw new UnsupportedException(); } @Override http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-hdfs/src/test/java/org/apache/tajo/storage/TestMergeScanner.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-hdfs/src/test/java/org/apache/tajo/storage/TestMergeScanner.java b/tajo-storage/tajo-storage-hdfs/src/test/java/org/apache/tajo/storage/TestMergeScanner.java index 331d3e8..35a5ea4 100644 --- a/tajo-storage/tajo-storage-hdfs/src/test/java/org/apache/tajo/storage/TestMergeScanner.java +++ b/tajo-storage/tajo-storage-hdfs/src/test/java/org/apache/tajo/storage/TestMergeScanner.java @@ -107,7 +107,7 @@ public class TestMergeScanner { KeyValueSet options = new KeyValueSet(); TableMeta meta = CatalogUtil.newTableMeta(storeType, options); - meta.setOptions(CatalogUtil.newPhysicalProperties(storeType)); + meta.setOptions(CatalogUtil.newDefaultProperty(storeType)); if (storeType.equalsIgnoreCase("AVRO")) { meta.putOption(StorageConstants.AVRO_SCHEMA_LITERAL, TEST_MULTIPLE_FILES_AVRO_SCHEMA); http://git-wip-us.apache.org/repos/asf/tajo/blob/8f3215d1/tajo-storage/tajo-storage-hdfs/src/test/java/org/apache/tajo/storage/TestStorages.java ---------------------------------------------------------------------- diff --git a/tajo-storage/tajo-storage-hdfs/src/test/java/org/apache/tajo/storage/TestStorages.java b/tajo-storage/tajo-storage-hdfs/src/test/java/org/apache/tajo/storage/TestStorages.java index dbfdac3..b53dbec 100644 --- a/tajo-storage/tajo-storage-hdfs/src/test/java/org/apache/tajo/storage/TestStorages.java +++ b/tajo-storage/tajo-storage-hdfs/src/test/java/org/apache/tajo/storage/TestStorages.java @@ -264,7 +264,7 @@ public class TestStorages { schema.addColumn("score", Type.FLOAT4); TableMeta meta = CatalogUtil.newTableMeta(storeType); - meta.setOptions(CatalogUtil.newPhysicalProperties(storeType)); + meta.setOptions(CatalogUtil.newDefaultProperty(storeType)); if (storeType.equalsIgnoreCase("AVRO")) { meta.putOption(StorageConstants.AVRO_SCHEMA_LITERAL, TEST_PROJECTION_AVRO_SCHEMA); @@ -341,7 +341,7 @@ public class TestStorages { KeyValueSet options = new KeyValueSet(); TableMeta meta = CatalogUtil.newTableMeta(storeType, options); - meta.setOptions(CatalogUtil.newPhysicalProperties(storeType)); + meta.setOptions(CatalogUtil.newDefaultProperty(storeType)); if (storeType.equalsIgnoreCase("AVRO")) { String path = FileUtil.getResourcePath("dataset/testVariousTypes.avsc").toString(); meta.putOption(StorageConstants.AVRO_SCHEMA_URL, path); @@ -414,7 +414,7 @@ public class TestStorages { KeyValueSet options = new KeyValueSet(); TableMeta meta = CatalogUtil.newTableMeta(storeType, options); - meta.setOptions(CatalogUtil.newPhysicalProperties(storeType)); + meta.setOptions(CatalogUtil.newDefaultProperty(storeType)); meta.putOption(StorageConstants.TEXT_NULL, "\\\\N"); meta.putOption(StorageConstants.RCFILE_NULL, "\\\\N"); meta.putOption(StorageConstants.RCFILE_SERDE, TextSerializerDeserializer.class.getName()); @@ -969,7 +969,7 @@ public class TestStorages { KeyValueSet options = new KeyValueSet(); TableMeta meta = CatalogUtil.newTableMeta(storeType, options); - meta.setOptions(CatalogUtil.newPhysicalProperties(storeType)); + meta.setOptions(CatalogUtil.newDefaultProperty(storeType)); Path tablePath = new Path(testDir, "testLessThanSchemaSize.data"); FileTablespace sm = TablespaceManager.getLocalFs(); @@ -1033,7 +1033,7 @@ public class TestStorages { KeyValueSet options = new KeyValueSet(); TableMeta meta = CatalogUtil.newTableMeta(storeType, options); - meta.setOptions(CatalogUtil.newPhysicalProperties(storeType)); + meta.setOptions(CatalogUtil.newDefaultProperty(storeType)); Path tablePath = new Path(testDir, "test_storetype_oversize.data"); FileTablespace sm = TablespaceManager.getLocalFs();
