Support Massin UDF
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/4adea167 Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/4adea167 Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/4adea167 Branch: refs/heads/master Commit: 4adea1677504e6dccdd7775d93df4625964f62e5 Parents: dfab0c1 Author: Hongbin Ma <mahong...@apache.org> Authored: Mon Mar 21 13:43:32 2016 +0800 Committer: Hongbin Ma <mahong...@apache.org> Committed: Thu Mar 24 14:45:02 2016 +0800 ---------------------------------------------------------------------- .../kylin/common/persistence/ResourceStore.java | 1 + .../kylin/common/restclient/Broadcaster.java | 3 +- .../common/restclient/SingleValueCache.java | 1 - .../kylin/cube/CubeCapabilityChecker.java | 32 ++-- .../kylin/cube/gridtable/CubeCodeSystem.java | 9 + .../cube/gridtable/TrimmedCubeCodeSystem.java | 139 ++++++--------- .../kylin/cube/gridtable/TrimmedDimEnc.java | 68 +++++++ .../gridtable/TrimmedDimensionSerializer.java | 57 ++++++ .../apache/kylin/cube/model/RowKeyColDesc.java | 1 - .../java/org/apache/kylin/gridtable/GTInfo.java | 1 + .../kylin/gridtable/GTSampleCodeSystem.java | 6 + .../java/org/apache/kylin/gridtable/GTUtil.java | 2 +- .../apache/kylin/gridtable/IGTCodeSystem.java | 3 + .../kylin/dict/BuildInFunctionTransformer.java | 175 +++++++++++++++++++ .../dict/TupleFilterFunctionTransformer.java | 172 ------------------ .../apache/kylin/dimension/FixedLenDimEnc.java | 20 +++ .../apache/kylin/metadata/MetadataManager.java | 48 ++++- .../filter/BuildInFunctionTupleFilter.java | 167 ++++++++++++++++++ .../metadata/filter/CompareTupleFilter.java | 3 +- .../metadata/filter/FunctionTupleFilter.java | 150 +--------------- .../kylin/metadata/filter/TupleFilter.java | 6 +- .../metadata/filter/TupleFilterSerializer.java | 6 +- .../metadata/filter/UDF/MassInTupleFilter.java | 140 +++++++++++++++ .../filter/UDF/MassInValueProvider.java | 25 +++ .../filter/UDF/MassInValueProviderFactory.java | 27 +++ .../metadata/filter/function/Functions.java | 60 +++++++ .../metadata/model/ExternalFilterDesc.java | 107 ++++++++++++ .../kylin/metadata/model/ISourceAware.java | 1 + .../kylin/metadata/realization/SQLDigest.java | 1 - .../localmeta/ext_filter/vip_customers.json | 6 + .../apache/kylin/query/ITKylinQueryTest.java | 2 +- .../apache/kylin/query/ITMassInQueryTest.java | 134 ++++++++++++++ .../src/test/resources/query/sql/query98.sql | 21 +++ .../test/resources/query/sql_massin/query01.sql | 30 ++++ .../test/resources/query/sql_massin/query02.sql | 30 ++++ .../test/resources/query/sql_massin/query03.sql | 30 ++++ .../test/resources/query/sql_massin/query04.sql | 27 +++ .../kylin/query/relnode/OLAPFilterRel.java | 6 +- .../kylin/query/schema/OLAPSchemaFactory.java | 18 +- .../org/apache/kylin/query/udf/MassInUDF.java | 28 +++ .../apache/kylin/rest/service/CacheService.java | 7 + .../common/coprocessor/FilterDecorator.java | 4 +- .../hbase/cube/v2/CubeHBaseEndpointRPC.java | 9 +- .../hbase/cube/v2/CubeSegmentScanner.java | 15 +- .../coprocessor/endpoint/CubeVisitService.java | 25 ++- .../filter/MassInValueProviderFactoryImpl.java | 43 +++++ .../cube/v2/filter/MassInValueProviderImpl.java | 80 +++++++++ 47 files changed, 1500 insertions(+), 446 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-common/src/main/java/org/apache/kylin/common/persistence/ResourceStore.java ---------------------------------------------------------------------- diff --git a/core-common/src/main/java/org/apache/kylin/common/persistence/ResourceStore.java b/core-common/src/main/java/org/apache/kylin/common/persistence/ResourceStore.java index 746527d..b07458c 100644 --- a/core-common/src/main/java/org/apache/kylin/common/persistence/ResourceStore.java +++ b/core-common/src/main/java/org/apache/kylin/common/persistence/ResourceStore.java @@ -51,6 +51,7 @@ abstract public class ResourceStore { public static final String SNAPSHOT_RESOURCE_ROOT = "/table_snapshot"; public static final String TABLE_EXD_RESOURCE_ROOT = "/table_exd"; public static final String TABLE_RESOURCE_ROOT = "/table"; + public static final String EXTERNAL_FILTER_RESOURCE_ROOT = "/ext_filter"; public static final String HYBRID_RESOURCE_ROOT = "/hybrid"; public static final String EXECUTE_RESOURCE_ROOT = "/execute"; public static final String EXECUTE_OUTPUT_RESOURCE_ROOT = "/execute_output"; http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-common/src/main/java/org/apache/kylin/common/restclient/Broadcaster.java ---------------------------------------------------------------------- diff --git a/core-common/src/main/java/org/apache/kylin/common/restclient/Broadcaster.java b/core-common/src/main/java/org/apache/kylin/common/restclient/Broadcaster.java index 871d77c..f8cd3be 100644 --- a/core-common/src/main/java/org/apache/kylin/common/restclient/Broadcaster.java +++ b/core-common/src/main/java/org/apache/kylin/common/restclient/Broadcaster.java @@ -143,6 +143,7 @@ public class Broadcaster { } public enum EVENT { + CREATE("create"), UPDATE("update"), DROP("drop"); private String text; @@ -166,7 +167,7 @@ public class Broadcaster { } public enum TYPE { - ALL("all"), CUBE("cube"), STREAMING("streaming"), KAFKA("kafka"), CUBE_DESC("cube_desc"), PROJECT("project"), INVERTED_INDEX("inverted_index"), INVERTED_INDEX_DESC("ii_desc"), TABLE("table"), DATA_MODEL("data_model"), HYBRID("hybrid"); + ALL("all"), CUBE("cube"), STREAMING("streaming"), KAFKA("kafka"), CUBE_DESC("cube_desc"), PROJECT("project"), INVERTED_INDEX("inverted_index"), INVERTED_INDEX_DESC("ii_desc"), TABLE("table"), DATA_MODEL("data_model"), EXTERNAL_FILTER("external_filter"), HYBRID("hybrid"); private String text; TYPE(String text) { http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-common/src/main/java/org/apache/kylin/common/restclient/SingleValueCache.java ---------------------------------------------------------------------- diff --git a/core-common/src/main/java/org/apache/kylin/common/restclient/SingleValueCache.java b/core-common/src/main/java/org/apache/kylin/common/restclient/SingleValueCache.java index 9acfeca..3631662 100644 --- a/core-common/src/main/java/org/apache/kylin/common/restclient/SingleValueCache.java +++ b/core-common/src/main/java/org/apache/kylin/common/restclient/SingleValueCache.java @@ -29,7 +29,6 @@ import org.apache.kylin.common.KylinConfig; /** * @author xjiang - * */ public abstract class SingleValueCache<K, V> extends AbstractRestCache<K, V> { http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-cube/src/main/java/org/apache/kylin/cube/CubeCapabilityChecker.java ---------------------------------------------------------------------- diff --git a/core-cube/src/main/java/org/apache/kylin/cube/CubeCapabilityChecker.java b/core-cube/src/main/java/org/apache/kylin/cube/CubeCapabilityChecker.java index 418b522..e21dc2b 100644 --- a/core-cube/src/main/java/org/apache/kylin/cube/CubeCapabilityChecker.java +++ b/core-cube/src/main/java/org/apache/kylin/cube/CubeCapabilityChecker.java @@ -29,7 +29,9 @@ import org.apache.kylin.cube.model.CubeDesc; import org.apache.kylin.cube.model.DimensionDesc; import org.apache.kylin.measure.MeasureType; import org.apache.kylin.measure.basic.BasicMeasureType; +import org.apache.kylin.metadata.filter.UDF.MassInTupleFilter; import org.apache.kylin.metadata.model.FunctionDesc; +import org.apache.kylin.metadata.model.IStorageAware; import org.apache.kylin.metadata.model.JoinDesc; import org.apache.kylin.metadata.model.MeasureDesc; import org.apache.kylin.metadata.model.TblColRef; @@ -62,28 +64,34 @@ public class CubeCapabilityChecker { Collection<FunctionDesc> aggrFunctions = digest.aggregations; Collection<TblColRef> unmatchedDimensions = unmatchedDimensions(dimensionColumns, cube); Collection<FunctionDesc> unmatchedAggregations = unmatchedAggregations(aggrFunctions, cube); - + // try custom measure types // in RAW query, unmatchedDimensions and unmatchedAggregations will null, so can't chose RAW cube well! -// if (!unmatchedDimensions.isEmpty() || !unmatchedAggregations.isEmpty()) { - tryCustomMeasureTypes(unmatchedDimensions, unmatchedAggregations, digest, cube, result); -// } - + // if (!unmatchedDimensions.isEmpty() || !unmatchedAggregations.isEmpty()) { + tryCustomMeasureTypes(unmatchedDimensions, unmatchedAggregations, digest, cube, result); + // } + // try dimension-as-measure if (!unmatchedAggregations.isEmpty()) { tryDimensionAsMeasures(unmatchedAggregations, digest, cube, result); } - + if (!unmatchedDimensions.isEmpty()) { logger.info("Exclude cube " + cube.getName() + " because unmatched dimensions"); return result; } - + if (!unmatchedAggregations.isEmpty()) { logger.info("Exclude cube " + cube.getName() + " because unmatched aggregations"); return result; } + if (cube.getStorageType() == IStorageAware.ID_HBASE && MassInTupleFilter.constainsMassInTupleFilter(digest.filter)) { + logger.info("Exclude cube " + cube.getName() + " because only v2 storage + v2 query engine supports massin"); + return result; + + } + // cost will be minded by caller result.capable = true; return result; @@ -155,7 +163,7 @@ public class CubeCapabilityChecker { Iterator<FunctionDesc> it = unmatchedAggregations.iterator(); while (it.hasNext()) { FunctionDesc functionDesc = it.next(); - + if (cubeFuncs.contains(functionDesc)) { it.remove(); continue; @@ -181,13 +189,13 @@ public class CubeCapabilityChecker { private static void tryCustomMeasureTypes(Collection<TblColRef> unmatchedDimensions, Collection<FunctionDesc> unmatchedAggregations, SQLDigest digest, CubeInstance cube, CapabilityResult result) { CubeDesc cubeDesc = cube.getDescriptor(); for (MeasureDesc measure : cubeDesc.getMeasures()) { -// if (unmatchedDimensions.isEmpty() && unmatchedAggregations.isEmpty()) -// break; - + // if (unmatchedDimensions.isEmpty() && unmatchedAggregations.isEmpty()) + // break; + MeasureType<?> measureType = measure.getFunction().getMeasureType(); if (measureType instanceof BasicMeasureType) continue; - + CapabilityInfluence inf = measureType.influenceCapabilityCheck(unmatchedDimensions, unmatchedAggregations, digest, measure); if (inf != null) result.influences.add(inf); http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeCodeSystem.java ---------------------------------------------------------------------- diff --git a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeCodeSystem.java b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeCodeSystem.java index e0dc4dd..97d1e85 100644 --- a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeCodeSystem.java +++ b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeCodeSystem.java @@ -98,6 +98,15 @@ public class CubeCodeSystem implements IGTCodeSystem { } @Override + public DimensionEncoding getDimEnc(int col) { + if (col < dimEncs.length) { + return dimEncs[col]; + } else { + return null; + } + } + + @Override public void encodeColumnValue(int col, Object value, ByteBuffer buf) { encodeColumnValue(col, value, 0, buf); } http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedCubeCodeSystem.java ---------------------------------------------------------------------- diff --git a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedCubeCodeSystem.java b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedCubeCodeSystem.java index b892520..cb3c55a 100644 --- a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedCubeCodeSystem.java +++ b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedCubeCodeSystem.java @@ -20,15 +20,18 @@ package org.apache.kylin.cube.gridtable; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; import java.io.IOException; -import java.io.ObjectInput; -import java.io.ObjectOutput; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; import java.nio.ByteBuffer; import java.util.Map; import org.apache.kylin.common.util.BytesSerializer; import org.apache.kylin.common.util.BytesUtil; import org.apache.kylin.common.util.ImmutableBitSet; +import org.apache.kylin.dimension.DictionaryDimEnc; import org.apache.kylin.dimension.DimensionEncoding; import org.apache.kylin.gridtable.DefaultGTComparator; import org.apache.kylin.gridtable.GTInfo; @@ -40,8 +43,7 @@ import org.apache.kylin.metadata.datatype.DataTypeSerializer; import com.google.common.collect.Maps; /** - * A limited code system where dimension value ser/des is disabled. - * Used inside coprocessor only. Because dictionary is not available. + * A limited code system which trims DictionaryDimEnc to TrimmedDimEnc (to avoid pushing down the useless dictionary) */ @SuppressWarnings({ "rawtypes", "unchecked" }) public class TrimmedCubeCodeSystem implements IGTCodeSystem { @@ -69,8 +71,8 @@ public class TrimmedCubeCodeSystem implements IGTCodeSystem { // for dimensions if (dimEnc != null) { - // use trimmed serializer cause no dictionary in coprocessor - serializers[i] = new TrimmedDimensionSerializer(dimEnc.getLengthOfEncoding()); + + serializers[i] = dimEnc.asDataTypeSerializer(); } // for measures else { @@ -95,6 +97,15 @@ public class TrimmedCubeCodeSystem implements IGTCodeSystem { } @Override + public DimensionEncoding getDimEnc(int col) { + if (col < dimEncs.length) { + return dimEncs[col]; + } else { + return null; + } + } + + @Override public void encodeColumnValue(int col, Object value, ByteBuffer buf) { encodeColumnValue(col, value, 0, buf); } @@ -139,6 +150,43 @@ public class TrimmedCubeCodeSystem implements IGTCodeSystem { return result; } + private static void writeDimensionEncoding(DimensionEncoding encoding, ByteBuffer out) { + try { + if (encoding == null) { + BytesUtil.writeVInt(1, out); + } else { + BytesUtil.writeVInt(0, out); + + if (encoding instanceof DictionaryDimEnc) { + encoding = new TrimmedDimEnc(encoding.getLengthOfEncoding()); + } + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(baos); + oos.writeObject(encoding); + BytesUtil.writeByteArray(baos.toByteArray(), out); + } + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + private static DimensionEncoding readDimensionEncoding(ByteBuffer in) { + try { + int isNull = BytesUtil.readVInt(in); + if (isNull == 1) { + return null; + } + + byte[] bytes = BytesUtil.readByteArray(in); + ByteArrayInputStream bais = new ByteArrayInputStream(bytes); + ObjectInputStream ois = new ObjectInputStream(bais); + DimensionEncoding ret = (DimensionEncoding) ois.readObject(); + return ret; + } catch (IOException | ClassNotFoundException e) { + throw new RuntimeException(e); + } + } + public static final BytesSerializer<TrimmedCubeCodeSystem> serializer = new BytesSerializer<TrimmedCubeCodeSystem>() { @Override public void serialize(TrimmedCubeCodeSystem value, ByteBuffer out) { @@ -151,7 +199,8 @@ public class TrimmedCubeCodeSystem implements IGTCodeSystem { BytesUtil.writeVInt(value.dimEncs.length, out); for (int i = 0; i < value.dimEncs.length; i++) { DimensionEncoding enc = value.dimEncs[i]; - BytesUtil.writeVInt(enc == null ? 0 : enc.getLengthOfEncoding(), out); + + writeDimensionEncoding(enc, out); } } @@ -168,85 +217,11 @@ public class TrimmedCubeCodeSystem implements IGTCodeSystem { DimensionEncoding[] dimEncs = new DimensionEncoding[BytesUtil.readVInt(in)]; for (int i = 0; i < dimEncs.length; i++) { - int fixedLen = BytesUtil.readVInt(in); - if (fixedLen > 0) - dimEncs[i] = new TrimmedDimEnc(fixedLen); + dimEncs[i] = readDimensionEncoding(in); } return new TrimmedCubeCodeSystem(dimEncs, dependentMetricsMap); } }; - static class TrimmedDimEnc extends DimensionEncoding { - final int fixedLen; - - TrimmedDimEnc(int fixedLen) { - this.fixedLen = fixedLen; - } - - @Override - public int getLengthOfEncoding() { - return fixedLen; - } - - @Override - public void encode(byte[] value, int valueLen, byte[] output, int outputOffset) { - throw new UnsupportedOperationException(); - } - - @Override - public String decode(byte[] bytes, int offset, int len) { - throw new UnsupportedOperationException(); - } - - @Override - public DataTypeSerializer<Object> asDataTypeSerializer() { - throw new UnsupportedOperationException(); - } - - @Override - public void writeExternal(ObjectOutput out) throws IOException { - throw new UnsupportedOperationException(); - } - - @Override - public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { - throw new UnsupportedOperationException(); - } - } - - static class TrimmedDimensionSerializer extends DataTypeSerializer<Object> { - - final int fixedLen; - - public TrimmedDimensionSerializer(int fixedLen) { - this.fixedLen = fixedLen; - } - - @Override - public int peekLength(ByteBuffer in) { - return fixedLen; - } - - @Override - public int maxLength() { - return fixedLen; - } - - @Override - public int getStorageBytesEstimate() { - return fixedLen; - } - - @Override - public void serialize(Object value, ByteBuffer out) { - throw new UnsupportedOperationException(); - } - - @Override - public Object deserialize(ByteBuffer in) { - throw new UnsupportedOperationException(); - } - } - } http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedDimEnc.java ---------------------------------------------------------------------- diff --git a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedDimEnc.java b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedDimEnc.java new file mode 100644 index 0000000..0350eb1 --- /dev/null +++ b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedDimEnc.java @@ -0,0 +1,68 @@ +/* + * 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.kylin.cube.gridtable; + +import org.apache.kylin.dimension.DimensionEncoding; +import org.apache.kylin.metadata.datatype.DataTypeSerializer; + +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; + +public class TrimmedDimEnc extends DimensionEncoding { + int fixedLen; + + //no-arg constructor is required for Externalizable + public TrimmedDimEnc() { + } + + public TrimmedDimEnc(int fixedLen) { + this.fixedLen = fixedLen; + } + + @Override + public int getLengthOfEncoding() { + return fixedLen; + } + + @Override + public void encode(byte[] value, int valueLen, byte[] output, int outputOffset) { + throw new UnsupportedOperationException(); + } + + @Override + public String decode(byte[] bytes, int offset, int len) { + throw new UnsupportedOperationException(); + } + + @Override + public DataTypeSerializer<Object> asDataTypeSerializer() { + return new TrimmedDimensionSerializer(fixedLen); + } + + @Override + public void writeExternal(ObjectOutput out) throws IOException { + out.writeShort(fixedLen); + } + + @Override + public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { + fixedLen = in.readShort(); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedDimensionSerializer.java ---------------------------------------------------------------------- diff --git a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedDimensionSerializer.java b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedDimensionSerializer.java new file mode 100644 index 0000000..05ef3e8 --- /dev/null +++ b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedDimensionSerializer.java @@ -0,0 +1,57 @@ +/* + * 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.kylin.cube.gridtable; + +import org.apache.kylin.metadata.datatype.DataTypeSerializer; + +import java.nio.ByteBuffer; + +public class TrimmedDimensionSerializer extends DataTypeSerializer<Object> { + + final int fixedLen; + + public TrimmedDimensionSerializer(int fixedLen) { + this.fixedLen = fixedLen; + } + + @Override + public int peekLength(ByteBuffer in) { + return fixedLen; + } + + @Override + public int maxLength() { + return fixedLen; + } + + @Override + public int getStorageBytesEstimate() { + return fixedLen; + } + + @Override + public void serialize(Object value, ByteBuffer out) { + throw new UnsupportedOperationException(); + } + + @Override + public Object deserialize(ByteBuffer in) { + throw new UnsupportedOperationException(); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyColDesc.java ---------------------------------------------------------------------- diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyColDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyColDesc.java index e72011c..08b1813 100644 --- a/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyColDesc.java +++ b/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyColDesc.java @@ -44,7 +44,6 @@ public class RowKeyColDesc { @JsonProperty("isUHC") private boolean isUHC;//is ultra high cardinality column - // computed private String encodingName; private String[] encodingArgs; http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-cube/src/main/java/org/apache/kylin/gridtable/GTInfo.java ---------------------------------------------------------------------- diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTInfo.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTInfo.java index 55356f9..80a5a55 100644 --- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTInfo.java +++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTInfo.java @@ -58,6 +58,7 @@ public class GTInfo { private GTInfo() { } + public String getTableName() { return tableName; } http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-cube/src/main/java/org/apache/kylin/gridtable/GTSampleCodeSystem.java ---------------------------------------------------------------------- diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTSampleCodeSystem.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTSampleCodeSystem.java index eb232e7..e379c42 100644 --- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTSampleCodeSystem.java +++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTSampleCodeSystem.java @@ -22,6 +22,7 @@ import java.nio.ByteBuffer; import org.apache.kylin.common.util.BytesSerializer; import org.apache.kylin.common.util.ImmutableBitSet; +import org.apache.kylin.dimension.DimensionEncoding; import org.apache.kylin.measure.MeasureAggregator; import org.apache.kylin.metadata.datatype.DataTypeSerializer; @@ -66,6 +67,11 @@ public class GTSampleCodeSystem implements IGTCodeSystem { } @Override + public DimensionEncoding getDimEnc(int col) { + return null; + } + + @Override public IGTComparator getComparator() { return comparator; } http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-cube/src/main/java/org/apache/kylin/gridtable/GTUtil.java ---------------------------------------------------------------------- diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTUtil.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTUtil.java index b231d18..08905ad 100644 --- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTUtil.java +++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTUtil.java @@ -88,7 +88,7 @@ public class GTUtil { } // shortcut for unEvaluatable filter - if (filter.isEvaluable() == false) { + if (!filter.isEvaluable()) { TupleFilter.collectColumns(filter, unevaluatableColumnCollector); return ConstantTupleFilter.TRUE; } http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-cube/src/main/java/org/apache/kylin/gridtable/IGTCodeSystem.java ---------------------------------------------------------------------- diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/IGTCodeSystem.java b/core-cube/src/main/java/org/apache/kylin/gridtable/IGTCodeSystem.java index dbd5e41..d75b158 100644 --- a/core-cube/src/main/java/org/apache/kylin/gridtable/IGTCodeSystem.java +++ b/core-cube/src/main/java/org/apache/kylin/gridtable/IGTCodeSystem.java @@ -21,6 +21,7 @@ package org.apache.kylin.gridtable; import java.nio.ByteBuffer; import org.apache.kylin.common.util.ImmutableBitSet; +import org.apache.kylin.dimension.DimensionEncoding; import org.apache.kylin.measure.MeasureAggregator; public interface IGTCodeSystem { @@ -34,6 +35,8 @@ public interface IGTCodeSystem { /** Return the max possible length of a column */ int maxCodeLength(int col); + + DimensionEncoding getDimEnc(int col); /** * Encode a value into code. http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-dictionary/src/main/java/org/apache/kylin/dict/BuildInFunctionTransformer.java ---------------------------------------------------------------------- diff --git a/core-dictionary/src/main/java/org/apache/kylin/dict/BuildInFunctionTransformer.java b/core-dictionary/src/main/java/org/apache/kylin/dict/BuildInFunctionTransformer.java new file mode 100644 index 0000000..ab5f703 --- /dev/null +++ b/core-dictionary/src/main/java/org/apache/kylin/dict/BuildInFunctionTransformer.java @@ -0,0 +1,175 @@ +/* + * 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.kylin.dict; + +import java.util.Collection; +import java.util.ListIterator; + +import org.apache.kylin.dimension.Dictionary; +import org.apache.kylin.dimension.IDimensionEncodingMap; +import org.apache.kylin.metadata.filter.BuildInFunctionTupleFilter; +import org.apache.kylin.metadata.filter.ColumnTupleFilter; +import org.apache.kylin.metadata.filter.CompareTupleFilter; +import org.apache.kylin.metadata.filter.ConstantTupleFilter; +import org.apache.kylin.metadata.filter.ITupleFilterTransformer; +import org.apache.kylin.metadata.filter.LogicalTupleFilter; +import org.apache.kylin.metadata.filter.TupleFilter; +import org.apache.kylin.metadata.filter.TupleFilter.FilterOperatorEnum; +import org.apache.kylin.metadata.model.TblColRef; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.collect.Lists; +import com.google.common.primitives.Primitives; + +/** + * only take effect when the compare filter has function + */ +public class BuildInFunctionTransformer implements ITupleFilterTransformer { + public static final Logger logger = LoggerFactory.getLogger(BuildInFunctionTransformer.class); + + private IDimensionEncodingMap dimEncMap; + + public BuildInFunctionTransformer(IDimensionEncodingMap dimEncMap) { + this.dimEncMap = dimEncMap; + } + + @Override + public TupleFilter transform(TupleFilter tupleFilter) { + TupleFilter translated = null; + if (tupleFilter instanceof CompareTupleFilter) { + //normal case + translated = translateCompareTupleFilter((CompareTupleFilter) tupleFilter); + if (translated != null) { + logger.info("Translated {" + tupleFilter + "} to IN clause: {" + translated + "}"); + } + } else if (tupleFilter instanceof BuildInFunctionTupleFilter) { + //like case + translated = translateFunctionTupleFilter((BuildInFunctionTupleFilter) tupleFilter); + if (translated != null) { + logger.info("Translated {" + tupleFilter + "} to IN clause: {" + translated + "}"); + } + } else if (tupleFilter instanceof LogicalTupleFilter) { + @SuppressWarnings("unchecked") + ListIterator<TupleFilter> childIterator = (ListIterator<TupleFilter>) tupleFilter.getChildren().listIterator(); + while (childIterator.hasNext()) { + TupleFilter transformed = transform(childIterator.next()); + if (transformed != null) + childIterator.set(transformed); + } + } + return translated == null ? tupleFilter : translated; + } + + private TupleFilter translateFunctionTupleFilter(BuildInFunctionTupleFilter buildInFunctionTupleFilter) { + if (!buildInFunctionTupleFilter.isValid()) + return null; + + TblColRef columnRef = buildInFunctionTupleFilter.getColumn(); + Dictionary<?> dict = dimEncMap.getDictionary(columnRef); + if (dict == null) + return null; + + CompareTupleFilter translated = new CompareTupleFilter(FilterOperatorEnum.IN); + translated.addChild(new ColumnTupleFilter(columnRef)); + + try { + for (int i = dict.getMinId(); i <= dict.getMaxId(); i++) { + Object dictVal = dict.getValueFromId(i); + if ((Boolean) buildInFunctionTupleFilter.invokeFunction(dictVal)) { + translated.addChild(new ConstantTupleFilter(dictVal)); + } + } + } catch (Exception e) { + logger.debug(e.getMessage()); + return null; + } + return translated; + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + private TupleFilter translateCompareTupleFilter(CompareTupleFilter compTupleFilter) { + if (compTupleFilter.getFunction() == null || (!(compTupleFilter.getFunction() instanceof BuildInFunctionTupleFilter))) + return null; + + BuildInFunctionTupleFilter buildInFunctionTupleFilter = (BuildInFunctionTupleFilter) compTupleFilter.getFunction(); + + if (!buildInFunctionTupleFilter.isValid()) + return null; + + TblColRef columnRef = buildInFunctionTupleFilter.getColumn(); + Dictionary<?> dict = dimEncMap.getDictionary(columnRef); + if (dict == null) + return null; + + CompareTupleFilter translated = new CompareTupleFilter(FilterOperatorEnum.IN); + translated.addChild(new ColumnTupleFilter(columnRef)); + + try { + Collection<Object> inValues = Lists.newArrayList(); + for (int i = dict.getMinId(); i <= dict.getMaxId(); i++) { + Object dictVal = dict.getValueFromId(i); + Object computedVal = buildInFunctionTupleFilter.invokeFunction(dictVal); + Class clazz = Primitives.wrap(computedVal.getClass()); + Object targetVal = compTupleFilter.getFirstValue(); + if (Primitives.isWrapperType(clazz)) + targetVal = clazz.cast(clazz.getDeclaredMethod("valueOf", String.class).invoke(null, compTupleFilter.getFirstValue())); + + int comp = ((Comparable) computedVal).compareTo(targetVal); + boolean compResult = false; + switch (compTupleFilter.getOperator()) { + case EQ: + compResult = comp == 0; + break; + case NEQ: + compResult = comp != 0; + break; + case LT: + compResult = comp < 0; + break; + case LTE: + compResult = comp <= 0; + break; + case GT: + compResult = comp > 0; + break; + case GTE: + compResult = comp >= 0; + break; + case IN: + compResult = compTupleFilter.getValues().contains(computedVal.toString()); + break; + case NOTIN: + compResult = !compTupleFilter.getValues().contains(computedVal.toString()); + break; + default: + break; + } + if (compResult) { + inValues.add(dictVal); + } + } + translated.addChild(new ConstantTupleFilter(inValues)); + } catch (Exception e) { + logger.debug(e.getMessage()); + return null; + } + return translated; + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-dictionary/src/main/java/org/apache/kylin/dict/TupleFilterFunctionTransformer.java ---------------------------------------------------------------------- diff --git a/core-dictionary/src/main/java/org/apache/kylin/dict/TupleFilterFunctionTransformer.java b/core-dictionary/src/main/java/org/apache/kylin/dict/TupleFilterFunctionTransformer.java deleted file mode 100644 index a5b4a7f..0000000 --- a/core-dictionary/src/main/java/org/apache/kylin/dict/TupleFilterFunctionTransformer.java +++ /dev/null @@ -1,172 +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.kylin.dict; - -import java.util.Collection; -import java.util.ListIterator; - -import org.apache.kylin.dimension.Dictionary; -import org.apache.kylin.dimension.IDimensionEncodingMap; -import org.apache.kylin.metadata.filter.ColumnTupleFilter; -import org.apache.kylin.metadata.filter.CompareTupleFilter; -import org.apache.kylin.metadata.filter.ConstantTupleFilter; -import org.apache.kylin.metadata.filter.FunctionTupleFilter; -import org.apache.kylin.metadata.filter.ITupleFilterTransformer; -import org.apache.kylin.metadata.filter.LogicalTupleFilter; -import org.apache.kylin.metadata.filter.TupleFilter; -import org.apache.kylin.metadata.filter.TupleFilter.FilterOperatorEnum; -import org.apache.kylin.metadata.model.TblColRef; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.google.common.collect.Lists; -import com.google.common.primitives.Primitives; - -/** - * only take effect when the compare filter has function - */ -public class TupleFilterFunctionTransformer implements ITupleFilterTransformer { - public static final Logger logger = LoggerFactory.getLogger(TupleFilterFunctionTransformer.class); - - private IDimensionEncodingMap dimEncMap; - - public TupleFilterFunctionTransformer(IDimensionEncodingMap dimEncMap) { - this.dimEncMap = dimEncMap; - } - - @Override - public TupleFilter transform(TupleFilter tupleFilter) { - TupleFilter translated = null; - if (tupleFilter instanceof CompareTupleFilter) { - translated = translateCompareTupleFilter((CompareTupleFilter) tupleFilter); - if (translated != null) { - logger.info("Translated {" + tupleFilter + "} to IN clause: {" + translated + "}"); - } - } else if (tupleFilter instanceof FunctionTupleFilter) { - translated = translateFunctionTupleFilter((FunctionTupleFilter) tupleFilter); - if (translated != null) { - logger.info("Translated {" + tupleFilter + "} to IN clause: {" + translated + "}"); - } - } else if (tupleFilter instanceof LogicalTupleFilter) { - @SuppressWarnings("unchecked") - ListIterator<TupleFilter> childIterator = (ListIterator<TupleFilter>) tupleFilter.getChildren().listIterator(); - while (childIterator.hasNext()) { - TupleFilter transformed = transform(childIterator.next()); - if (transformed != null) - childIterator.set(transformed); - } - } - return translated == null ? tupleFilter : translated; - } - - private TupleFilter translateFunctionTupleFilter(FunctionTupleFilter functionTupleFilter) { - if (!functionTupleFilter.isValid()) - return null; - - TblColRef columnRef = functionTupleFilter.getColumn(); - Dictionary<?> dict = dimEncMap.getDictionary(columnRef); - if (dict == null) - return null; - - CompareTupleFilter translated = new CompareTupleFilter(FilterOperatorEnum.IN); - translated.addChild(new ColumnTupleFilter(columnRef)); - - try { - for (int i = dict.getMinId(); i <= dict.getMaxId(); i++) { - Object dictVal = dict.getValueFromId(i); - if ((Boolean) functionTupleFilter.invokeFunction(dictVal)) { - translated.addChild(new ConstantTupleFilter(dictVal)); - } - } - } catch (Exception e) { - logger.debug(e.getMessage()); - return null; - } - return translated; - } - - @SuppressWarnings({ "unchecked", "rawtypes" }) - private TupleFilter translateCompareTupleFilter(CompareTupleFilter compTupleFilter) { - if (compTupleFilter.getFunction() == null) - return null; - - FunctionTupleFilter functionTupleFilter = compTupleFilter.getFunction(); - if (!functionTupleFilter.isValid()) - return null; - - TblColRef columnRef = functionTupleFilter.getColumn(); - Dictionary<?> dict = dimEncMap.getDictionary(columnRef); - if (dict == null) - return null; - - CompareTupleFilter translated = new CompareTupleFilter(FilterOperatorEnum.IN); - translated.addChild(new ColumnTupleFilter(columnRef)); - - try { - Collection<Object> inValues = Lists.newArrayList(); - for (int i = dict.getMinId(); i <= dict.getMaxId(); i++) { - Object dictVal = dict.getValueFromId(i); - Object computedVal = functionTupleFilter.invokeFunction(dictVal); - Class clazz = Primitives.wrap(computedVal.getClass()); - Object targetVal = compTupleFilter.getFirstValue(); - if (Primitives.isWrapperType(clazz)) - targetVal = clazz.cast(clazz.getDeclaredMethod("valueOf", String.class).invoke(null, compTupleFilter.getFirstValue())); - - int comp = ((Comparable) computedVal).compareTo(targetVal); - boolean compResult = false; - switch (compTupleFilter.getOperator()) { - case EQ: - compResult = comp == 0; - break; - case NEQ: - compResult = comp != 0; - break; - case LT: - compResult = comp < 0; - break; - case LTE: - compResult = comp <= 0; - break; - case GT: - compResult = comp > 0; - break; - case GTE: - compResult = comp >= 0; - break; - case IN: - compResult = compTupleFilter.getValues().contains(computedVal.toString()); - break; - case NOTIN: - compResult = !compTupleFilter.getValues().contains(computedVal.toString()); - break; - default: - break; - } - if (compResult) { - inValues.add(dictVal); - } - } - translated.addChild(new ConstantTupleFilter(inValues)); - } catch (Exception e) { - logger.debug(e.getMessage()); - return null; - } - return translated; - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-metadata/src/main/java/org/apache/kylin/dimension/FixedLenDimEnc.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/dimension/FixedLenDimEnc.java b/core-metadata/src/main/java/org/apache/kylin/dimension/FixedLenDimEnc.java index 195bdb9..9d99d62 100644 --- a/core-metadata/src/main/java/org/apache/kylin/dimension/FixedLenDimEnc.java +++ b/core-metadata/src/main/java/org/apache/kylin/dimension/FixedLenDimEnc.java @@ -59,11 +59,31 @@ public class FixedLenDimEnc extends DimensionEncoding { transient private int avoidVerbose = 0; + //no-arg constructor is required for Externalizable + public FixedLenDimEnc() { + } + public FixedLenDimEnc(int len) { this.fixedLen = len; } @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + FixedLenDimEnc that = (FixedLenDimEnc) o; + + return fixedLen == that.fixedLen; + + } + + @Override + public int hashCode() { + return fixedLen; + } + + @Override public int getLengthOfEncoding() { return fixedLen; } http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-metadata/src/main/java/org/apache/kylin/metadata/MetadataManager.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/MetadataManager.java b/core-metadata/src/main/java/org/apache/kylin/metadata/MetadataManager.java index 3aa3810..e7b345f 100644 --- a/core-metadata/src/main/java/org/apache/kylin/metadata/MetadataManager.java +++ b/core-metadata/src/main/java/org/apache/kylin/metadata/MetadataManager.java @@ -41,6 +41,7 @@ import org.apache.kylin.common.restclient.Broadcaster; import org.apache.kylin.common.restclient.CaseInsensitiveStringCache; import org.apache.kylin.common.util.JsonUtil; import org.apache.kylin.metadata.model.DataModelDesc; +import org.apache.kylin.metadata.model.ExternalFilterDesc; import org.apache.kylin.metadata.model.TableDesc; import org.apache.kylin.metadata.project.ProjectInstance; import org.apache.kylin.metadata.project.ProjectManager; @@ -67,6 +68,7 @@ public class MetadataManager { public static final Serializer<TableDesc> TABLE_SERIALIZER = new JsonSerializer<TableDesc>(TableDesc.class); public static final Serializer<DataModelDesc> MODELDESC_SERIALIZER = new JsonSerializer<DataModelDesc>(DataModelDesc.class); + public static final Serializer<ExternalFilterDesc> EXTERNAL_FILTER_DESC_SERIALIZER = new JsonSerializer<ExternalFilterDesc>(ExternalFilterDesc.class); // static cached instances private static final ConcurrentHashMap<KylinConfig, MetadataManager> CACHE = new ConcurrentHashMap<KylinConfig, MetadataManager>(); @@ -109,15 +111,13 @@ public class MetadataManager { private CaseInsensitiveStringCache<Map<String, String>> srcTableExdMap; // name => DataModelDesc private CaseInsensitiveStringCache<DataModelDesc> dataModelDescMap; + // name => External Filter Desc + private CaseInsensitiveStringCache<ExternalFilterDesc> extFilterMap; private MetadataManager(KylinConfig config) throws IOException { init(config); } - public static String concatDataModelResourcePath(String modelName) { - return ResourceStore.DATA_MODEL_DESC_RESOURCE_ROOT + "/" + modelName + MetadataConstants.FILE_SURFIX; - } - /** * Tell MetadataManager that the instance has changed. The cube info will * be stored Reload the cube desc and source table A broadcast must be sent @@ -165,6 +165,11 @@ public class MetadataManager { return result; } + public ExternalFilterDesc getExtFilterDesc(String filterTableName) { + ExternalFilterDesc result = extFilterMap.get(filterTableName); + return result; + } + /** * Get table extended info. Keys are defined in {@link MetadataConstants} * @@ -209,13 +214,15 @@ public class MetadataManager { private void init(KylinConfig config) throws IOException { this.config = config; - this.srcTableMap = new CaseInsensitiveStringCache<TableDesc>(config, Broadcaster.TYPE.TABLE); - this.srcTableExdMap = new CaseInsensitiveStringCache<Map<String, String>>(config, Broadcaster.TYPE.TABLE); - this.dataModelDescMap = new CaseInsensitiveStringCache<DataModelDesc>(config, Broadcaster.TYPE.DATA_MODEL); + this.srcTableMap = new CaseInsensitiveStringCache<>(config, Broadcaster.TYPE.TABLE); + this.srcTableExdMap = new CaseInsensitiveStringCache<>(config, Broadcaster.TYPE.TABLE); + this.dataModelDescMap = new CaseInsensitiveStringCache<>(config, Broadcaster.TYPE.DATA_MODEL); + this.extFilterMap = new CaseInsensitiveStringCache<>(config, Broadcaster.TYPE.EXTERNAL_FILTER); reloadAllSourceTable(); reloadAllSourceTableExd(); reloadAllDataModel(); + reloadAllExternalFilter(); } private void reloadAllSourceTableExd() throws IOException { @@ -263,6 +270,20 @@ public class MetadataManager { return attrs; } + private void reloadAllExternalFilter() throws IOException { + ResourceStore store = getStore(); + logger.debug("Reloading ExternalFilter from folder " + store.getReadableResourcePath(ResourceStore.EXTERNAL_FILTER_RESOURCE_ROOT)); + + extFilterMap.clear(); + + List<String> paths = store.collectResourceRecursively(ResourceStore.EXTERNAL_FILTER_RESOURCE_ROOT, MetadataConstants.FILE_SURFIX); + for (String path : paths) { + reloadExternalFilterAt(path); + } + + logger.debug("Loaded " + extFilterMap.size() + " SourceTable(s)"); + } + private void reloadAllSourceTable() throws IOException { ResourceStore store = getStore(); logger.debug("Reloading SourceTable from folder " + store.getReadableResourcePath(ResourceStore.TABLE_RESOURCE_ROOT)); @@ -292,6 +313,19 @@ public class MetadataManager { return t; } + private ExternalFilterDesc reloadExternalFilterAt(String path) throws IOException { + ResourceStore store = getStore(); + ExternalFilterDesc t = store.getResource(path, ExternalFilterDesc.class, EXTERNAL_FILTER_DESC_SERIALIZER); + if (t == null) { + return null; + } + t.init(); + + extFilterMap.putLocal(t.getName(), t); + + return t; + } + public void reloadSourceTableExt(String tableIdentity) throws IOException { reloadSourceTableExdAt(TableDesc.concatExdResourcePath(tableIdentity)); } http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-metadata/src/main/java/org/apache/kylin/metadata/filter/BuildInFunctionTupleFilter.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/BuildInFunctionTupleFilter.java b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/BuildInFunctionTupleFilter.java new file mode 100644 index 0000000..9d283fd --- /dev/null +++ b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/BuildInFunctionTupleFilter.java @@ -0,0 +1,167 @@ +/* + * 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.kylin.metadata.filter; + +import java.io.Serializable; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.nio.ByteBuffer; +import java.util.Collection; +import java.util.List; + +import org.apache.kylin.common.util.BytesUtil; +import org.apache.kylin.metadata.filter.function.BuiltInMethod; +import org.apache.kylin.metadata.model.TblColRef; +import org.apache.kylin.metadata.tuple.IEvaluatableTuple; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.collect.Lists; +import com.google.common.primitives.Primitives; + +public class BuildInFunctionTupleFilter extends FunctionTupleFilter { + public static final Logger logger = LoggerFactory.getLogger(BuildInFunctionTupleFilter.class); + + private String name; + // FIXME Only supports single parameter functions currently + private TupleFilter columnContainerFilter;//might be a ColumnTupleFilter(simple case) or FunctionTupleFilter(complex case like substr(lower())) + private int colPosition; + private Method method; + private List<Serializable> methodParams; + private boolean isValid = false; + + + public BuildInFunctionTupleFilter(String name) { + super(Lists.<TupleFilter> newArrayList(), FilterOperatorEnum.FUNCTION); + this.methodParams = Lists.newArrayList(); + + if (name != null) { + this.name = name.toUpperCase(); + initMethod(); + } + } + + public String getName() { + return name; + } + + public TblColRef getColumn() { + if (columnContainerFilter == null) + return null; + + if (columnContainerFilter instanceof ColumnTupleFilter) + return ((ColumnTupleFilter) columnContainerFilter).getColumn(); + else if (columnContainerFilter instanceof BuildInFunctionTupleFilter) + return ((BuildInFunctionTupleFilter) columnContainerFilter).getColumn(); + + throw new UnsupportedOperationException("Wrong type TupleFilter in FunctionTupleFilter."); + } + + public Object invokeFunction(Object input) throws InvocationTargetException, IllegalAccessException { + if (columnContainerFilter instanceof ColumnTupleFilter) + methodParams.set(colPosition, (Serializable) input); + else if (columnContainerFilter instanceof BuildInFunctionTupleFilter) + methodParams.set(colPosition, (Serializable) ((BuildInFunctionTupleFilter) columnContainerFilter).invokeFunction(input)); + return method.invoke(null, (Object[]) (methodParams.toArray())); + } + + public boolean isValid() { + return isValid && method != null && methodParams.size() == children.size(); + } + + @Override + @SuppressWarnings("unchecked") + public void addChild(TupleFilter child) { + if (child instanceof ColumnTupleFilter || child instanceof BuildInFunctionTupleFilter) { + columnContainerFilter = child; + colPosition = methodParams.size(); + methodParams.add(null); + } else if (child instanceof ConstantTupleFilter) { + Serializable constVal = (Serializable) child.getValues().iterator().next(); + try { + Class<?> clazz = Primitives.wrap(method.getParameterTypes()[methodParams.size()]); + if (!Primitives.isWrapperType(clazz)) + methodParams.add(constVal); + else + methodParams.add((Serializable) clazz.cast(clazz.getDeclaredMethod("valueOf", String.class).invoke(null, constVal))); + } catch (Exception e) { + logger.warn(e.getMessage()); + isValid = false; + } + } + super.addChild(child); + } + + @Override + public boolean isEvaluable() { + return false; + } + + @Override + public boolean evaluate(IEvaluatableTuple tuple, IFilterCodeSystem<?> cs) { + throw new UnsupportedOperationException("Function filter cannot be evaluated immediately"); + } + + @Override + public Collection<String> getValues() { + throw new UnsupportedOperationException("Function filter cannot be evaluated immediately"); + } + + @Override + public void serialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) { + BytesUtil.writeUTFString(name, buffer); + BytesUtil.writeVInt(colPosition, buffer); + BytesUtil.writeVInt(isValid ? 1 : 0, buffer); + } + + @Override + public void deserialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) { + + this.name = BytesUtil.readUTFString(buffer); + this.initMethod(); + + this.colPosition = BytesUtil.readVInt(buffer); + this.isValid = BytesUtil.readVInt(buffer) == 1; + } + + @Override + public String toString() { + StringBuilder sb = new StringBuilder(); + sb.append(name); + sb.append("("); + for (int i = 0; i < methodParams.size(); i++) { + if (colPosition == i) { + sb.append(columnContainerFilter); + } else { + sb.append(methodParams.get(i)); + } + if (i < methodParams.size() - 1) + sb.append(","); + } + sb.append(")"); + return sb.toString(); + } + + private void initMethod() { + if (BuiltInMethod.MAP.containsKey(name)) { + this.method = BuiltInMethod.MAP.get(name).method; + isValid = true; + } + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-metadata/src/main/java/org/apache/kylin/metadata/filter/CompareTupleFilter.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/CompareTupleFilter.java b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/CompareTupleFilter.java index fc0bab7..7124fed 100644 --- a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/CompareTupleFilter.java +++ b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/CompareTupleFilter.java @@ -26,6 +26,7 @@ import java.util.Map; import java.util.Set; import org.apache.kylin.common.util.BytesUtil; +import org.apache.kylin.metadata.model.FunctionDesc; import org.apache.kylin.metadata.model.TblColRef; import org.apache.kylin.metadata.tuple.IEvaluatableTuple; @@ -150,7 +151,7 @@ public class CompareTupleFilter extends TupleFilter { // extract tuple value Object tupleValue = null; for (TupleFilter filter : this.children) { - if (isConstant(filter) == false) { + if (!isConstant(filter)) { filter.evaluate(tuple, cs); tupleValue = filter.getValues().iterator().next(); } http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-metadata/src/main/java/org/apache/kylin/metadata/filter/FunctionTupleFilter.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/FunctionTupleFilter.java b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/FunctionTupleFilter.java index 2a08728..72c247a 100644 --- a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/FunctionTupleFilter.java +++ b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/FunctionTupleFilter.java @@ -6,9 +6,9 @@ * 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. @@ -18,149 +18,11 @@ package org.apache.kylin.metadata.filter; -import java.io.Serializable; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; -import java.nio.ByteBuffer; -import java.util.Collection; import java.util.List; -import org.apache.kylin.common.util.BytesUtil; -import org.apache.kylin.metadata.filter.function.BuiltInMethod; -import org.apache.kylin.metadata.model.TblColRef; -import org.apache.kylin.metadata.tuple.IEvaluatableTuple; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +public abstract class FunctionTupleFilter extends TupleFilter { -import com.google.common.collect.Lists; -import com.google.common.primitives.Primitives; - -public class FunctionTupleFilter extends TupleFilter { - public static final Logger logger = LoggerFactory.getLogger(FunctionTupleFilter.class); - - private String name; - // FIXME Only supports single parameter functions currently - private TupleFilter columnContainerFilter; - private int colPosition; - private Method method; - private List<Serializable> methodParams; - private boolean isValid = false; - - public FunctionTupleFilter(String name) { - super(Lists.<TupleFilter> newArrayList(), FilterOperatorEnum.FUNCTION); - this.methodParams = Lists.newArrayList(); - - if (name != null) { - this.name = name.toUpperCase(); - initMethod(); - } - } - - public String getName() { - return name; - } - - public TblColRef getColumn() { - if (columnContainerFilter == null) - return null; - - if (columnContainerFilter instanceof ColumnTupleFilter) - return ((ColumnTupleFilter) columnContainerFilter).getColumn(); - else if (columnContainerFilter instanceof FunctionTupleFilter) - return ((FunctionTupleFilter) columnContainerFilter).getColumn(); - - throw new UnsupportedOperationException("Wrong type TupleFilter in FunctionTupleFilter."); - } - - public Object invokeFunction(Object input) throws InvocationTargetException, IllegalAccessException { - if (columnContainerFilter instanceof ColumnTupleFilter) - methodParams.set(colPosition, (Serializable) input); - else if (columnContainerFilter instanceof FunctionTupleFilter) - methodParams.set(colPosition, (Serializable) ((FunctionTupleFilter) columnContainerFilter).invokeFunction(input)); - return method.invoke(null, (Object[]) (methodParams.toArray())); - } - - public boolean isValid() { - return isValid && method != null && methodParams.size() == children.size(); - } - - @Override - @SuppressWarnings("unchecked") - public void addChild(TupleFilter child) { - if (child instanceof ColumnTupleFilter || child instanceof FunctionTupleFilter) { - columnContainerFilter = child; - colPosition = methodParams.size(); - methodParams.add(null); - } else if (child instanceof ConstantTupleFilter) { - Serializable constVal = (Serializable) child.getValues().iterator().next(); - try { - Class<?> clazz = Primitives.wrap(method.getParameterTypes()[methodParams.size()]); - if (!Primitives.isWrapperType(clazz)) - methodParams.add(constVal); - else - methodParams.add((Serializable) clazz.cast(clazz.getDeclaredMethod("valueOf", String.class).invoke(null, constVal))); - } catch (Exception e) { - logger.warn(e.getMessage()); - isValid = false; - } - } - super.addChild(child); - } - - @Override - public boolean isEvaluable() { - return false; - } - - @Override - public boolean evaluate(IEvaluatableTuple tuple, IFilterCodeSystem<?> cs) { - throw new UnsupportedOperationException("Function filter cannot be evaluated immediately"); - } - - @Override - public Collection<String> getValues() { - throw new UnsupportedOperationException("Function filter cannot be evaluated immediately"); - } - - @Override - void serialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) { - BytesUtil.writeUTFString(name, buffer); - BytesUtil.writeVInt(colPosition, buffer); - BytesUtil.writeVInt(isValid ? 1 : 0, buffer); - } - - @Override - public void deserialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) { - - this.name = BytesUtil.readUTFString(buffer); - this.initMethod(); - - this.colPosition = BytesUtil.readVInt(buffer); - this.isValid = BytesUtil.readVInt(buffer) == 1; - } - - @Override - public String toString() { - StringBuilder sb = new StringBuilder(); - sb.append(name); - sb.append("("); - for (int i = 0; i < methodParams.size(); i++) { - if (colPosition == i) { - sb.append(columnContainerFilter); - } else { - sb.append(methodParams.get(i)); - } - if (i < methodParams.size() - 1) - sb.append(","); - } - sb.append(")"); - return sb.toString(); - } - - private void initMethod() { - if (BuiltInMethod.MAP.containsKey(name)) { - this.method = BuiltInMethod.MAP.get(name).method; - isValid = true; - } + protected FunctionTupleFilter(List<TupleFilter> filters, FilterOperatorEnum op) { + super(filters, op); } -} \ No newline at end of file +} http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-metadata/src/main/java/org/apache/kylin/metadata/filter/TupleFilter.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/TupleFilter.java b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/TupleFilter.java index 1e23499..bc5a284 100644 --- a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/TupleFilter.java +++ b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/TupleFilter.java @@ -38,7 +38,7 @@ import com.google.common.collect.Maps; public abstract class TupleFilter { public enum FilterOperatorEnum { - EQ(1), NEQ(2), GT(3), LT(4), GTE(5), LTE(6), ISNULL(7), ISNOTNULL(8), IN(9), NOTIN(10), AND(20), OR(21), NOT(22), COLUMN(30), CONSTANT(31), DYNAMIC(32), EXTRACT(33), CASE(34), FUNCTION(35); + EQ(1), NEQ(2), GT(3), LT(4), GTE(5), LTE(6), ISNULL(7), ISNOTNULL(8), IN(9), NOTIN(10), AND(20), OR(21), NOT(22), COLUMN(30), CONSTANT(31), DYNAMIC(32), EXTRACT(33), CASE(34), FUNCTION(35),MASSIN(36); private final int value; @@ -205,9 +205,9 @@ public abstract class TupleFilter { public abstract Collection<?> getValues(); - abstract void serialize(IFilterCodeSystem<?> cs,ByteBuffer buffer); + public abstract void serialize(IFilterCodeSystem<?> cs, ByteBuffer buffer); - abstract void deserialize(IFilterCodeSystem<?> cs,ByteBuffer buffer); + public abstract void deserialize(IFilterCodeSystem<?> cs, ByteBuffer buffer); public static boolean isEvaluableRecursively(TupleFilter filter) { if (filter == null) http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-metadata/src/main/java/org/apache/kylin/metadata/filter/TupleFilterSerializer.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/TupleFilterSerializer.java b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/TupleFilterSerializer.java index 39ccb15..bcb005f 100644 --- a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/TupleFilterSerializer.java +++ b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/TupleFilterSerializer.java @@ -25,6 +25,7 @@ import java.util.Map; import java.util.Stack; import org.apache.kylin.common.util.BytesUtil; +import org.apache.kylin.metadata.filter.UDF.MassInTupleFilter; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -185,7 +186,10 @@ public class TupleFilterSerializer { filter = new DynamicTupleFilter(null); break; case FUNCTION: - filter = new FunctionTupleFilter(null); + filter = new BuildInFunctionTupleFilter(null); + break; + case MASSIN: + filter = new MassInTupleFilter(); break; default: throw new IllegalStateException("Error FilterOperatorEnum: " + op.getValue()); http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-metadata/src/main/java/org/apache/kylin/metadata/filter/UDF/MassInTupleFilter.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/UDF/MassInTupleFilter.java b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/UDF/MassInTupleFilter.java new file mode 100644 index 0000000..0cb416c --- /dev/null +++ b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/UDF/MassInTupleFilter.java @@ -0,0 +1,140 @@ +/* + * 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.kylin.metadata.filter.UDF; + +import java.nio.ByteBuffer; +import java.util.Collection; + +import org.apache.kylin.common.KylinConfig; +import org.apache.kylin.common.util.BytesUtil; +import org.apache.kylin.metadata.MetadataManager; +import org.apache.kylin.metadata.filter.ColumnTupleFilter; +import org.apache.kylin.metadata.filter.ConstantTupleFilter; +import org.apache.kylin.metadata.filter.FunctionTupleFilter; +import org.apache.kylin.metadata.filter.IFilterCodeSystem; +import org.apache.kylin.metadata.filter.TupleFilter; +import org.apache.kylin.metadata.filter.function.Functions; +import org.apache.kylin.metadata.model.ExternalFilterDesc; +import org.apache.kylin.metadata.model.TblColRef; +import org.apache.kylin.metadata.tuple.IEvaluatableTuple; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.base.Preconditions; +import com.google.common.collect.Lists; + +public class MassInTupleFilter extends FunctionTupleFilter { + + public static final Logger logger = LoggerFactory.getLogger(MassInTupleFilter.class); + public static MassInValueProviderFactory VALUE_PROVIDER_FACTORY = null; + + private transient MassInValueProvider valueProvider = null; + private transient TblColRef column; + + private String filterTableName;//key in MetadataManager.extFilterMap + private String filterTableResourceIdentifier;//HDFS path, or hbase table name depending on FilterTableType + private Functions.FilterTableType filterTableType; + + public MassInTupleFilter() { + super(Lists.<TupleFilter> newArrayList(), TupleFilter.FilterOperatorEnum.MASSIN); + } + + @Override + public boolean evaluate(IEvaluatableTuple tuple, IFilterCodeSystem<?> cs) { + Preconditions.checkNotNull(tuple); + Preconditions.checkNotNull(column); + + Object colValue = tuple.getValue(column); + + if (valueProvider == null) { + valueProvider = VALUE_PROVIDER_FACTORY.getProvider(filterTableType, filterTableResourceIdentifier, column); + } + boolean ret = valueProvider.getMassInValues().contains(colValue); + return ret; + } + + @Override + public Collection<?> getValues() { + return null; + } + + @Override + public boolean isEvaluable() { + return true; + } + + @Override + public void addChild(TupleFilter child) { + if (child instanceof ColumnTupleFilter) { + super.addChild(child); + ColumnTupleFilter columnFilter = (ColumnTupleFilter) child; + if (this.column != null) { + throw new IllegalStateException("Duplicate columns! old is " + column.getName() + " and new is " + columnFilter.getColumn().getName()); + } + this.column = columnFilter.getColumn(); + + } else if (child instanceof ConstantTupleFilter) { + // super.addChild(child) is omitted because the filter table name is useless at storage side, + // we'll extract the useful filterTableResourceIdentifier,filterTableType etc and save it at the MassInTupleFilter itself + + if (filterTableName == null) { + filterTableName = (String) child.getValues().iterator().next(); + ExternalFilterDesc externalFilterDesc = MetadataManager.getInstance(KylinConfig.getInstanceFromEnv()).getExtFilterDesc(filterTableName); + if (externalFilterDesc == null) { + throw new IllegalArgumentException("External filter named " + filterTableName + " is not found"); + } + filterTableType = externalFilterDesc.getFilterTableType(); + filterTableResourceIdentifier = externalFilterDesc.getFilterResourceIdentifier(); + } + } else { + throw new IllegalStateException("MassInTupleFilter only has two children: one ColumnTupleFilter and one ConstantTupleFilter"); + } + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + @Override + public void serialize(IFilterCodeSystem cs, ByteBuffer buffer) { + BytesUtil.writeUTFString(filterTableName, buffer); + BytesUtil.writeUTFString(filterTableResourceIdentifier, buffer); + BytesUtil.writeUTFString(filterTableType.toString(), buffer); + } + + @Override + public void deserialize(IFilterCodeSystem<?> cs, ByteBuffer buffer) { + filterTableName = BytesUtil.readUTFString(buffer); + filterTableResourceIdentifier = BytesUtil.readUTFString(buffer); + filterTableType = Functions.FilterTableType.valueOf(BytesUtil.readUTFString(buffer)); + } + + public static boolean constainsMassInTupleFilter(TupleFilter filter) { + if (filter == null) + return false; + + if (filter instanceof MassInTupleFilter) { + return true; + } + + for (TupleFilter child : filter.getChildren()) { + if (constainsMassInTupleFilter(child)) + return true; + } + return false; + } + +} http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-metadata/src/main/java/org/apache/kylin/metadata/filter/UDF/MassInValueProvider.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/UDF/MassInValueProvider.java b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/UDF/MassInValueProvider.java new file mode 100644 index 0000000..d834331 --- /dev/null +++ b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/UDF/MassInValueProvider.java @@ -0,0 +1,25 @@ +/* + * 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.kylin.metadata.filter.UDF; + +import java.util.Set; + +public interface MassInValueProvider { + Set<?> getMassInValues(); +} http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-metadata/src/main/java/org/apache/kylin/metadata/filter/UDF/MassInValueProviderFactory.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/UDF/MassInValueProviderFactory.java b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/UDF/MassInValueProviderFactory.java new file mode 100644 index 0000000..0ae7e6a --- /dev/null +++ b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/UDF/MassInValueProviderFactory.java @@ -0,0 +1,27 @@ +/* + * 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.kylin.metadata.filter.UDF; + +import org.apache.kylin.dimension.DimensionEncoding; +import org.apache.kylin.metadata.filter.function.Functions; +import org.apache.kylin.metadata.model.TblColRef; + +public interface MassInValueProviderFactory { + MassInValueProvider getProvider(Functions.FilterTableType filterTableType, String filterResourceIdentifier, TblColRef col); +} http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-metadata/src/main/java/org/apache/kylin/metadata/filter/function/Functions.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/filter/function/Functions.java b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/function/Functions.java new file mode 100644 index 0000000..7931437 --- /dev/null +++ b/core-metadata/src/main/java/org/apache/kylin/metadata/filter/function/Functions.java @@ -0,0 +1,60 @@ +/* + * 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.kylin.metadata.filter.function; + +import java.lang.reflect.InvocationTargetException; +import java.util.Map; + +import org.apache.kylin.metadata.filter.BuildInFunctionTupleFilter; +import org.apache.kylin.metadata.filter.TupleFilter; +import org.apache.kylin.metadata.filter.UDF.MassInTupleFilter; + +import com.google.common.collect.Maps; + +public class Functions { + + public enum FilterTableType { + HDFS, HBASE_TABLE + } + + private static Map<String, Class> SUPPORTED_UDF = Maps.newHashMap(); + + static { + SUPPORTED_UDF.put("MASSIN", MassInTupleFilter.class); + } + + public static TupleFilter getFunctionTupleFilter(String name) { + if (name == null) { + throw new IllegalStateException("Function name cannot be null"); + } + + name = name.toUpperCase(); + + if (SUPPORTED_UDF.containsKey(name)) { + try { + return (TupleFilter) SUPPORTED_UDF.get(name).getConstructor().newInstance(); + } catch (InstantiationException | IllegalAccessException | InvocationTargetException | NoSuchMethodException e) { + throw new RuntimeException("Failed to on constructing FunctionTupleFilter for " + name); + } + } + + return new BuildInFunctionTupleFilter(name); + + } +} http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-metadata/src/main/java/org/apache/kylin/metadata/model/ExternalFilterDesc.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/model/ExternalFilterDesc.java b/core-metadata/src/main/java/org/apache/kylin/metadata/model/ExternalFilterDesc.java new file mode 100644 index 0000000..cf53aef --- /dev/null +++ b/core-metadata/src/main/java/org/apache/kylin/metadata/model/ExternalFilterDesc.java @@ -0,0 +1,107 @@ +/* + * 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.kylin.metadata.model; + +import org.apache.kylin.common.persistence.ResourceStore; +import org.apache.kylin.common.persistence.RootPersistentEntity; +import org.apache.kylin.metadata.filter.function.Functions; + +import com.fasterxml.jackson.annotation.JsonAutoDetect; +import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** + * External filter enables user to register dynamic external filters out the scope of cubes. + * External filters are maintained logically in a filter store (which may or may not share same physical store with cubes), + * and are accessed by each cube shard at runtime. + * + * Currently the way to use external filter is 1. register external filter through REST 2. use UDF to specify conditions on external filter + */ +@SuppressWarnings("serial") +@JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE) +public class ExternalFilterDesc extends RootPersistentEntity implements ISourceAware { + + @JsonProperty("name") + private String name; + @JsonProperty("filter_resource_identifier") + private String filterResourceIdentifier; + @JsonProperty("filter_table_type") + private Functions.FilterTableType filterTableType; + @JsonProperty("source_type") + private int sourceType = ISourceAware.ID_EXTERNAL; + + public String getResourcePath() { + return concatResourcePath(getName()); + } + + public static String concatResourcePath(String name) { + return ResourceStore.TABLE_EXD_RESOURCE_ROOT + "/" + name + ".json"; + } + + // ============================================================================ + + + public String getFilterResourceIdentifier() { + return filterResourceIdentifier; + } + + public void setFilterResourceIdentifier(String filterResourceIdentifier) { + this.filterResourceIdentifier = filterResourceIdentifier; + } + + public Functions.FilterTableType getFilterTableType() { + return filterTableType; + } + + public void setFilterTableType(Functions.FilterTableType filterTableType) { + this.filterTableType = filterTableType; + } + + public String getName() { + return this.name; + } + + public void setName(String name) { + this.name = name; + } + + public void init() { + } + + @Override + public String toString() { + return "ExternalFilterDesc [ name=" + name + " filter table resource identifier " + this.filterResourceIdentifier + "]"; + } + + /** create a mockup table for unit test */ + public static ExternalFilterDesc mockup(String tableName) { + ExternalFilterDesc mockup = new ExternalFilterDesc(); + mockup.setName(tableName); + return mockup; + } + + @Override + public int getSourceType() { + return sourceType; + } + + public void setSourceType(int sourceType) { + this.sourceType = sourceType; + } +} http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-metadata/src/main/java/org/apache/kylin/metadata/model/ISourceAware.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/model/ISourceAware.java b/core-metadata/src/main/java/org/apache/kylin/metadata/model/ISourceAware.java index 8cfda15..0f98d5d 100644 --- a/core-metadata/src/main/java/org/apache/kylin/metadata/model/ISourceAware.java +++ b/core-metadata/src/main/java/org/apache/kylin/metadata/model/ISourceAware.java @@ -23,6 +23,7 @@ public interface ISourceAware { public static final int ID_HIVE = 0; public static final int ID_STREAMING = 1; public static final int ID_SPARKSQL = 5; + public static final int ID_EXTERNAL = 7; int getSourceType(); } http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/core-metadata/src/main/java/org/apache/kylin/metadata/realization/SQLDigest.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/realization/SQLDigest.java b/core-metadata/src/main/java/org/apache/kylin/metadata/realization/SQLDigest.java index 08bfc8c..aa90fc5 100644 --- a/core-metadata/src/main/java/org/apache/kylin/metadata/realization/SQLDigest.java +++ b/core-metadata/src/main/java/org/apache/kylin/metadata/realization/SQLDigest.java @@ -19,7 +19,6 @@ package org.apache.kylin.metadata.realization; import java.util.Collection; -import java.util.List; import org.apache.kylin.metadata.filter.TupleFilter; import org.apache.kylin.metadata.model.FunctionDesc; http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/examples/test_case_data/localmeta/ext_filter/vip_customers.json ---------------------------------------------------------------------- diff --git a/examples/test_case_data/localmeta/ext_filter/vip_customers.json b/examples/test_case_data/localmeta/ext_filter/vip_customers.json new file mode 100644 index 0000000..49d54b9 --- /dev/null +++ b/examples/test_case_data/localmeta/ext_filter/vip_customers.json @@ -0,0 +1,6 @@ +{ + "name": "vip_customers", + "filter_table_type": "HDFS", + "filter_resource_identifier": "/tmp/vip_customers.txt", + "source_type": 7 +} \ No newline at end of file