You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by ma...@apache.org on 2016/03/24 07:45:35 UTC
[1/4] kylin git commit: Support Massin UDF
Repository: kylin
Updated Branches:
refs/heads/master 870322003 -> 4adea1677
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/kylin-it/src/test/java/org/apache/kylin/query/ITKylinQueryTest.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/query/ITKylinQueryTest.java b/kylin-it/src/test/java/org/apache/kylin/query/ITKylinQueryTest.java
index 4168df2..09fcf4b 100644
--- a/kylin-it/src/test/java/org/apache/kylin/query/ITKylinQueryTest.java
+++ b/kylin-it/src/test/java/org/apache/kylin/query/ITKylinQueryTest.java
@@ -96,7 +96,7 @@ public class ITKylinQueryTest extends KylinTestBase {
@Test
public void testSingleRunQuery() throws Exception {
- String queryFileName = "src/test/resources/query/sql_tableau/query20.sql";
+ String queryFileName = "src/test/resources/query/temp/query01.sql";
File sqlFile = new File(queryFileName);
if (sqlFile.exists()) {
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/kylin-it/src/test/java/org/apache/kylin/query/ITMassInQueryTest.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/query/ITMassInQueryTest.java b/kylin-it/src/test/java/org/apache/kylin/query/ITMassInQueryTest.java
new file mode 100644
index 0000000..a92e298
--- /dev/null
+++ b/kylin-it/src/test/java/org/apache/kylin/query/ITMassInQueryTest.java
@@ -0,0 +1,134 @@
+/*
+ * 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.query;
+
+import java.io.File;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.Random;
+import java.util.Set;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.kylin.engine.mr.HadoopUtil;
+import org.dbunit.Assertion;
+import org.dbunit.database.DatabaseConfig;
+import org.dbunit.database.DatabaseConnection;
+import org.dbunit.database.IDatabaseConnection;
+import org.dbunit.dataset.ITable;
+import org.dbunit.ext.h2.H2Connection;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.google.common.collect.Sets;
+
+/**
+ */
+public class ITMassInQueryTest extends KylinTestBase {
+
+ FileSystem fileSystem;
+ Set<Long> vipSellers;
+
+ @BeforeClass
+ public static void setUp() throws SQLException {
+ }
+
+ @AfterClass
+ public static void tearDown() {
+ }
+
+ @Before
+ public void setup() throws Exception {
+
+ ITKylinQueryTest.clean();
+ ITKylinQueryTest.joinType = "inner";
+ ITKylinQueryTest.setupAll();
+
+ Configuration hconf = HadoopUtil.getCurrentConfiguration();
+ fileSystem = FileSystem.get(hconf);
+
+ int sellerCount = 200;
+ Random r = new Random();
+ vipSellers = Sets.newHashSet();
+ for (int i = 0; i < sellerCount; i++) {
+ vipSellers.add(10000000L + r.nextInt(1500));
+ }
+
+ Path path = new Path("/tmp/vip_customers.txt");
+ fileSystem.delete(path, false);
+ FSDataOutputStream outputStream = fileSystem.create(path);
+ org.apache.commons.io.IOUtils.write(StringUtils.join(vipSellers, "\n"), outputStream);
+ outputStream.close();
+
+ System.out.println("The filter is " + vipSellers);
+ }
+
+ @After
+ public void after() throws Exception {
+ ITKylinQueryTest.clean();
+
+ }
+
+ @Test
+ public void testMassInQuery() throws Exception {
+ compare("src/test/resources/query/sql_massin", null, true);
+ }
+
+ protected void compare(String queryFolder, String[] exclusiveQuerys, boolean needSort) throws Exception {
+ printInfo("---------- test folder: " + queryFolder);
+ Set<String> exclusiveSet = buildExclusiveSet(exclusiveQuerys);
+
+ List<File> sqlFiles = getFilesFromFolder(new File(queryFolder), ".sql");
+ for (File sqlFile : sqlFiles) {
+ String queryName = StringUtils.split(sqlFile.getName(), '.')[0];
+ if (exclusiveSet.contains(queryName)) {
+ continue;
+ }
+ String sql = getTextFromFile(sqlFile);
+
+ // execute Kylin
+ printInfo("Query Result from Kylin - " + queryName + " (" + queryFolder + ")");
+ IDatabaseConnection kylinConn = new DatabaseConnection(cubeConnection);
+ ITable kylinTable = executeQuery(kylinConn, queryName, sql, needSort);
+
+ // execute H2
+ sql = sql.replace("massin(test_kylin_fact.SELLER_ID,'vip_customers')", "test_kylin_fact.SELLER_ID in ( " + org.apache.commons.lang.StringUtils.join(vipSellers, ",") + ")");
+ printInfo("Query Result from H2 - " + queryName);
+ printInfo("Query for H2 - " + sql);
+ H2Connection h2Conn = new H2Connection(h2Connection, null);
+ h2Conn.getConfig().setProperty(DatabaseConfig.PROPERTY_DATATYPE_FACTORY, new TestH2DataTypeFactory());
+ ITable h2Table = executeQuery(h2Conn, queryName, sql, needSort);
+
+ try {
+ // compare the result
+ Assertion.assertEquals(h2Table, kylinTable);
+ } catch (Throwable t) {
+ printInfo("execAndCompQuery failed on: " + sqlFile.getAbsolutePath());
+ throw t;
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/kylin-it/src/test/resources/query/sql/query98.sql
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/resources/query/sql/query98.sql b/kylin-it/src/test/resources/query/sql/query98.sql
new file mode 100644
index 0000000..d73a4cb
--- /dev/null
+++ b/kylin-it/src/test/resources/query/sql/query98.sql
@@ -0,0 +1,21 @@
+--
+-- 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.
+--
+
+select cal_dt, sum(price) as x,leaf_categ_id as GMV
+ from test_kylin_fact
+ group by leaf_categ_id, cal_dt
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/kylin-it/src/test/resources/query/sql_massin/query01.sql
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/resources/query/sql_massin/query01.sql b/kylin-it/src/test/resources/query/sql_massin/query01.sql
new file mode 100644
index 0000000..f21e5fb
--- /dev/null
+++ b/kylin-it/src/test/resources/query/sql_massin/query01.sql
@@ -0,0 +1,30 @@
+--
+-- 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.
+--
+
+SELECT
+ count(*) as TRANS_CNT
+ FROM test_kylin_fact
+ inner JOIN edw.test_cal_dt as test_cal_dt
+ ON test_kylin_fact.cal_dt = test_cal_dt.cal_dt
+ inner JOIN test_category_groupings
+ ON test_kylin_fact.leaf_categ_id = test_category_groupings.leaf_categ_id AND test_kylin_fact.lstg_site_id = test_category_groupings.site_id
+ inner JOIN edw.test_sites as test_sites
+ ON test_kylin_fact.lstg_site_id = test_sites.site_id
+ inner JOIN edw.test_seller_type_dim as test_seller_type_dim
+ ON test_kylin_fact.slr_segment_cd = test_seller_type_dim.seller_type_cd
+ where massin(test_kylin_fact.SELLER_ID,'vip_customers')
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/kylin-it/src/test/resources/query/sql_massin/query02.sql
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/resources/query/sql_massin/query02.sql b/kylin-it/src/test/resources/query/sql_massin/query02.sql
new file mode 100644
index 0000000..8981b75
--- /dev/null
+++ b/kylin-it/src/test/resources/query/sql_massin/query02.sql
@@ -0,0 +1,30 @@
+--
+-- 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.
+--
+
+
+ select test_kylin_fact.leaf_categ_id,sum(price) as GMV, count(*) as x
+ from test_kylin_fact inner JOIN edw.test_cal_dt as test_cal_dt
+ ON test_kylin_fact.cal_dt = test_cal_dt.cal_dt
+ inner JOIN test_category_groupings
+ ON test_kylin_fact.leaf_categ_id = test_category_groupings.leaf_categ_id AND test_kylin_fact.lstg_site_id = test_category_groupings.site_id
+ inner JOIN edw.test_sites as test_sites
+ ON test_kylin_fact.lstg_site_id = test_sites.site_id
+ inner JOIN edw.test_seller_type_dim as test_seller_type_dim
+ ON test_kylin_fact.slr_segment_cd = test_seller_type_dim.seller_type_cd
+ where massin(test_kylin_fact.SELLER_ID,'vip_customers')
+ group by test_kylin_fact.leaf_categ_id
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/kylin-it/src/test/resources/query/sql_massin/query03.sql
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/resources/query/sql_massin/query03.sql b/kylin-it/src/test/resources/query/sql_massin/query03.sql
new file mode 100644
index 0000000..27a5918
--- /dev/null
+++ b/kylin-it/src/test/resources/query/sql_massin/query03.sql
@@ -0,0 +1,30 @@
+--
+-- 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.
+--
+
+
+ select test_kylin_fact.lstg_format_name, sum(price) as GMV, count(*) as x
+ from test_kylin_fact inner JOIN edw.test_cal_dt as test_cal_dt
+ ON test_kylin_fact.cal_dt = test_cal_dt.cal_dt
+ inner JOIN test_category_groupings
+ ON test_kylin_fact.leaf_categ_id = test_category_groupings.leaf_categ_id AND test_kylin_fact.lstg_site_id = test_category_groupings.site_id
+ inner JOIN edw.test_sites as test_sites
+ ON test_kylin_fact.lstg_site_id = test_sites.site_id
+ inner JOIN edw.test_seller_type_dim as test_seller_type_dim
+ ON test_kylin_fact.slr_segment_cd = test_seller_type_dim.seller_type_cd
+ where massin(test_kylin_fact.SELLER_ID,'vip_customers')
+ group by test_kylin_fact.lstg_format_name having count(*) > 2
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/kylin-it/src/test/resources/query/sql_massin/query04.sql
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/resources/query/sql_massin/query04.sql b/kylin-it/src/test/resources/query/sql_massin/query04.sql
new file mode 100644
index 0000000..f11939c
--- /dev/null
+++ b/kylin-it/src/test/resources/query/sql_massin/query04.sql
@@ -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.
+--
+
+select sum(price) as GMV, count(*) as TRANS_CNT FROM test_kylin_fact
+ inner JOIN edw.test_cal_dt as test_cal_dt
+ ON test_kylin_fact.cal_dt = test_cal_dt.cal_dt
+ inner JOIN test_category_groupings
+ ON test_kylin_fact.leaf_categ_id = test_category_groupings.leaf_categ_id
+ AND test_kylin_fact.lstg_site_id = test_category_groupings.site_id
+where (test_kylin_fact.cal_dt < DATE '2012-05-01' or test_kylin_fact.cal_dt > DATE '2013-05-01')
+and massin(test_kylin_fact.SELLER_ID,'vip_customers')
+
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/query/src/main/java/org/apache/kylin/query/relnode/OLAPFilterRel.java
----------------------------------------------------------------------
diff --git a/query/src/main/java/org/apache/kylin/query/relnode/OLAPFilterRel.java b/query/src/main/java/org/apache/kylin/query/relnode/OLAPFilterRel.java
index 5ea138f..2806b4b 100644
--- a/query/src/main/java/org/apache/kylin/query/relnode/OLAPFilterRel.java
+++ b/query/src/main/java/org/apache/kylin/query/relnode/OLAPFilterRel.java
@@ -57,7 +57,7 @@ import org.apache.kylin.metadata.filter.CompareTupleFilter;
import org.apache.kylin.metadata.filter.ConstantTupleFilter;
import org.apache.kylin.metadata.filter.DynamicTupleFilter;
import org.apache.kylin.metadata.filter.ExtractTupleFilter;
-import org.apache.kylin.metadata.filter.FunctionTupleFilter;
+import org.apache.kylin.metadata.filter.function.Functions;
import org.apache.kylin.metadata.filter.LogicalTupleFilter;
import org.apache.kylin.metadata.filter.TupleFilter;
import org.apache.kylin.metadata.filter.TupleFilter.FilterOperatorEnum;
@@ -130,12 +130,12 @@ public class OLAPFilterRel extends Filter implements OLAPRel {
if (op.getName().equalsIgnoreCase("extract_date")) {
filter = new ExtractTupleFilter(FilterOperatorEnum.EXTRACT);
} else {
- filter = new FunctionTupleFilter(op.getName());
+ filter = Functions.getFunctionTupleFilter(op.getName());
}
break;
case LIKE:
case OTHER_FUNCTION:
- filter = new FunctionTupleFilter(op.getName());
+ filter = Functions.getFunctionTupleFilter(op.getName());
break;
default:
throw new UnsupportedOperationException(op.getName());
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/query/src/main/java/org/apache/kylin/query/schema/OLAPSchemaFactory.java
----------------------------------------------------------------------
diff --git a/query/src/main/java/org/apache/kylin/query/schema/OLAPSchemaFactory.java b/query/src/main/java/org/apache/kylin/query/schema/OLAPSchemaFactory.java
index f8711b1..44e9e47 100644
--- a/query/src/main/java/org/apache/kylin/query/schema/OLAPSchemaFactory.java
+++ b/query/src/main/java/org/apache/kylin/query/schema/OLAPSchemaFactory.java
@@ -29,15 +29,21 @@ import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.SchemaFactory;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.util.ConversionUtil;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang.StringUtils;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.metadata.model.DatabaseDesc;
import org.apache.kylin.metadata.model.TableDesc;
import org.apache.kylin.metadata.project.ProjectInstance;
import org.apache.kylin.metadata.project.ProjectManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
*/
public class OLAPSchemaFactory implements SchemaFactory {
+ public static final Logger logger = LoggerFactory.getLogger(OLAPSchemaFactory.class);
+
static {
/*
@@ -107,7 +113,13 @@ public class OLAPSchemaFactory implements SchemaFactory {
out.write(" \"factory\": \"org.apache.kylin.query.schema.OLAPSchemaFactory\",\n");
out.write(" \"operand\": {\n");
out.write(" \"" + SCHEMA_PROJECT + "\": \"" + project + "\"\n");
- out.write(" }\n");
+ out.write(" },\n");
+ out.write(" \"functions\": [\n");
+ out.write(" {\n");
+ out.write(" name: 'MASSIN',\n");
+ out.write(" className: 'org.apache.kylin.query.udf.MassInUDF'\n");
+ out.write(" }\n");
+ out.write(" ]\n");
out.write(" }\n");
if (++counter != schemaCounts.size()) {
@@ -118,8 +130,10 @@ public class OLAPSchemaFactory implements SchemaFactory {
out.write(" ]\n");
out.write("}\n");
out.close();
-
tmp.deleteOnExit();
+
+ logger.info("Schema json:" + StringUtils.join(FileUtils.readLines(tmp), "\n"));
+
return tmp;
} catch (IOException e) {
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/query/src/main/java/org/apache/kylin/query/udf/MassInUDF.java
----------------------------------------------------------------------
diff --git a/query/src/main/java/org/apache/kylin/query/udf/MassInUDF.java b/query/src/main/java/org/apache/kylin/query/udf/MassInUDF.java
new file mode 100644
index 0000000..596fa1e
--- /dev/null
+++ b/query/src/main/java/org/apache/kylin/query/udf/MassInUDF.java
@@ -0,0 +1,28 @@
+/*
+ * 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.query.udf;
+
+import org.apache.calcite.linq4j.function.Parameter;
+
+public class MassInUDF {
+
+ public boolean eval(@Parameter(name = "col") Object col, @Parameter(name = "filterTable") String filterTable) {
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/server/src/main/java/org/apache/kylin/rest/service/CacheService.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/kylin/rest/service/CacheService.java b/server/src/main/java/org/apache/kylin/rest/service/CacheService.java
index 9a6036d..4ec2116 100644
--- a/server/src/main/java/org/apache/kylin/rest/service/CacheService.java
+++ b/server/src/main/java/org/apache/kylin/rest/service/CacheService.java
@@ -200,6 +200,11 @@ public class CacheService extends BasicService {
IIDescManager.clearCache();
CubeDescManager.clearCache();
break;
+ case EXTERNAL_FILTER:
+ getMetadataManager().reloadTableCache(cacheKey);
+ IIDescManager.clearCache();
+ CubeDescManager.clearCache();
+ break;
case DATA_MODEL:
getMetadataManager().reloadDataModelDesc(cacheKey);
IIDescManager.clearCache();
@@ -262,6 +267,8 @@ public class CacheService extends BasicService {
break;
case TABLE:
throw new UnsupportedOperationException(log);
+ case EXTERNAL_FILTER:
+ throw new UnsupportedOperationException(log);
case DATA_MODEL:
getMetadataManager().removeModelCache(cacheKey);
break;
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/common/coprocessor/FilterDecorator.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/common/coprocessor/FilterDecorator.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/common/coprocessor/FilterDecorator.java
index a782020..f87bd30 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/common/coprocessor/FilterDecorator.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/common/coprocessor/FilterDecorator.java
@@ -24,7 +24,7 @@ import java.util.Set;
import org.apache.kylin.common.util.Bytes;
import org.apache.kylin.cube.kv.RowKeyColumnIO;
import org.apache.kylin.dict.DictCodeSystem;
-import org.apache.kylin.dict.TupleFilterFunctionTransformer;
+import org.apache.kylin.dict.BuildInFunctionTransformer;
import org.apache.kylin.dimension.Dictionary;
import org.apache.kylin.dimension.DimensionEncoding;
import org.apache.kylin.dimension.IDimensionEncodingMap;
@@ -151,7 +151,7 @@ public class FilterDecorator implements TupleFilterSerializer.Decorator {
if (filter == null)
return null;
- TupleFilterFunctionTransformer translator = new TupleFilterFunctionTransformer(dimEncMap);
+ BuildInFunctionTransformer translator = new BuildInFunctionTransformer(dimEncMap);
filter = translator.transform(filter);
// un-evaluatable filter is replaced with TRUE
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseEndpointRPC.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseEndpointRPC.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseEndpointRPC.java
index e6f9ac1..bde2196 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseEndpointRPC.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseEndpointRPC.java
@@ -135,7 +135,12 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
}
try {
current++;
- return queue.poll(timeout, TimeUnit.MILLISECONDS);
+ byte[] ret = queue.poll(timeout, TimeUnit.MILLISECONDS);
+ if (ret == null) {
+ throw new RuntimeException("Timeout visiting cube!");
+ } else {
+ return ret;
+ }
} catch (InterruptedException e) {
throw new RuntimeException("error when waiting queue", e);
}
@@ -236,7 +241,7 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
return Lists.newArrayList(Pair.newPair(getByteArrayForShort(baseShard), getByteArrayForShort((short) (baseShard + shardNum - 1))));
} else {
//0,1,2,3,4 wants 4,0
- return Lists.newArrayList(Pair.newPair(getByteArrayForShort(baseShard), getByteArrayForShort((short) (totalShards - 1))),//
+ return Lists.newArrayList(Pair.newPair(getByteArrayForShort(baseShard), getByteArrayForShort((short) (totalShards - 1))), //
Pair.newPair(getByteArrayForShort((short) 0), getByteArrayForShort((short) (baseShard + shardNum - totalShards - 1))));
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeSegmentScanner.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeSegmentScanner.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeSegmentScanner.java
index c69d31f..ea38508 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeSegmentScanner.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeSegmentScanner.java
@@ -40,7 +40,8 @@ import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.gridtable.CubeGridTable;
import org.apache.kylin.cube.gridtable.CuboidToGridTableMapping;
import org.apache.kylin.cube.model.CubeDesc;
-import org.apache.kylin.dict.TupleFilterFunctionTransformer;
+import org.apache.kylin.dict.BuildInFunctionTransformer;
+import org.apache.kylin.dimension.DimensionEncoding;
import org.apache.kylin.gridtable.EmptyGTScanner;
import org.apache.kylin.gridtable.GTInfo;
import org.apache.kylin.gridtable.GTRecord;
@@ -52,8 +53,10 @@ import org.apache.kylin.gridtable.IGTScanner;
import org.apache.kylin.metadata.datatype.DataType;
import org.apache.kylin.metadata.filter.ITupleFilterTransformer;
import org.apache.kylin.metadata.filter.TupleFilter;
+import org.apache.kylin.metadata.filter.UDF.MassInTupleFilter;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.TblColRef;
+import org.apache.kylin.storage.hbase.cube.v2.filter.MassInValueProviderFactoryImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -73,7 +76,7 @@ public class CubeSegmentScanner implements IGTScanner {
final Cuboid cuboid;
public CubeSegmentScanner(CubeSegment cubeSeg, Cuboid cuboid, Set<TblColRef> dimensions, Set<TblColRef> groups, //
- Collection<FunctionDesc> metrics, TupleFilter filter, boolean allowPreAggregate) {
+ Collection<FunctionDesc> metrics, TupleFilter filter, boolean allowPreAggregate) {
this.cuboid = cuboid;
this.cubeSeg = cubeSeg;
this.info = CubeGridTable.newGTInfo(cubeSeg, cuboid.getId());
@@ -81,7 +84,7 @@ public class CubeSegmentScanner implements IGTScanner {
CuboidToGridTableMapping mapping = cuboid.getCuboidToGridTableMapping();
// translate FunctionTupleFilter to IN clause
- ITupleFilterTransformer translator = new TupleFilterFunctionTransformer(cubeSeg.getDimensionEncodingMap());
+ ITupleFilterTransformer translator = new BuildInFunctionTransformer(cubeSeg.getDimensionEncodingMap());
filter = translator.transform(filter);
//replace the constant values in filter to dictionary codes
@@ -242,6 +245,12 @@ public class CubeSegmentScanner implements IGTScanner {
public Scanner() {
CubeHBaseRPC rpc;
if ("scan".equalsIgnoreCase(BackdoorToggles.getHbaseCubeQueryProtocol())) {
+ MassInTupleFilter.VALUE_PROVIDER_FACTORY = new MassInValueProviderFactoryImpl(new MassInValueProviderFactoryImpl.DimEncAware() {
+ @Override
+ public DimensionEncoding getDimEnc(TblColRef col) {
+ return info.getCodeSystem().getDimEnc(col.getColumnDesc().getZeroBasedIndex());
+ }
+ });
rpc = new CubeHBaseScanRPC(cubeSeg, cuboid, info);
} else {
rpc = new CubeHBaseEndpointRPC(cubeSeg, cuboid, info);//default behavior
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java
index 7ac4be1..9f42c1c 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java
@@ -40,10 +40,13 @@ import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.kylin.common.util.Bytes;
import org.apache.kylin.common.util.CompressionUtils;
import org.apache.kylin.cube.kv.RowConstants;
+import org.apache.kylin.dimension.DimensionEncoding;
import org.apache.kylin.gridtable.GTRecord;
import org.apache.kylin.gridtable.GTScanRequest;
import org.apache.kylin.gridtable.IGTScanner;
import org.apache.kylin.gridtable.IGTStore;
+import org.apache.kylin.metadata.filter.UDF.MassInTupleFilter;
+import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.storage.hbase.common.coprocessor.CoprocessorBehavior;
import org.apache.kylin.storage.hbase.cube.v2.CellListIterator;
import org.apache.kylin.storage.hbase.cube.v2.CubeHBaseRPC;
@@ -51,6 +54,7 @@ import org.apache.kylin.storage.hbase.cube.v2.HBaseReadonlyStore;
import org.apache.kylin.storage.hbase.cube.v2.RawScan;
import org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos;
import org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.IntList;
+import org.apache.kylin.storage.hbase.cube.v2.filter.MassInValueProviderFactoryImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -141,6 +145,7 @@ public class CubeVisitService extends CubeVisitProtos.CubeVisitService implement
@Override
public void visitCube(RpcController controller, CubeVisitProtos.CubeVisitRequest request, RpcCallback<CubeVisitProtos.CubeVisitResponse> done) {
+
RegionScanner innerScanner = null;
HRegion region = null;
@@ -153,8 +158,15 @@ public class CubeVisitService extends CubeVisitProtos.CubeVisitService implement
region = env.getRegion();
region.startRegionOperation();
- GTScanRequest scanReq = GTScanRequest.serializer.deserialize(ByteBuffer.wrap(HBaseZeroCopyByteString.zeroCopyGetBytes(request.getGtScanRequest())));
- RawScan hbaseRawScan = RawScan.serializer.deserialize(ByteBuffer.wrap(HBaseZeroCopyByteString.zeroCopyGetBytes(request.getHbaseRawScan())));
+ final GTScanRequest scanReq = GTScanRequest.serializer.deserialize(ByteBuffer.wrap(HBaseZeroCopyByteString.zeroCopyGetBytes(request.getGtScanRequest())));
+ final RawScan hbaseRawScan = RawScan.serializer.deserialize(ByteBuffer.wrap(HBaseZeroCopyByteString.zeroCopyGetBytes(request.getHbaseRawScan())));
+
+ MassInTupleFilter.VALUE_PROVIDER_FACTORY = new MassInValueProviderFactoryImpl(new MassInValueProviderFactoryImpl.DimEncAware() {
+ @Override
+ public DimensionEncoding getDimEnc(TblColRef col) {
+ return scanReq.getInfo().getCodeSystem().getDimEnc(col.getColumnDesc().getZeroBasedIndex());
+ }
+ });
List<List<Integer>> hbaseColumnsToGT = Lists.newArrayList();
for (IntList intList : request.getHbaseColumnsToGTList()) {
@@ -165,7 +177,7 @@ public class CubeVisitService extends CubeVisitProtos.CubeVisitService implement
//if has shard, fill region shard to raw scan start/end
updateRawScanByCurrentRegion(hbaseRawScan, region, request.getRowkeyPreambleSize() - RowConstants.ROWKEY_CUBOIDID_LEN);
}
-
+
Scan scan = CubeHBaseRPC.buildScan(hbaseRawScan);
appendProfileInfo(sb);
@@ -181,8 +193,8 @@ public class CubeVisitService extends CubeVisitProtos.CubeVisitService implement
IGTStore store = new HBaseReadonlyStore(cellListIterator, scanReq, hbaseRawScan.hbaseColumns, hbaseColumnsToGT, request.getRowkeyPreambleSize());
IGTScanner rawScanner = store.scan(scanReq);
- IGTScanner finalScanner = scanReq.decorateScanner(rawScanner,//
- behavior.ordinal() >= CoprocessorBehavior.SCAN_FILTER.ordinal(),//
+ IGTScanner finalScanner = scanReq.decorateScanner(rawScanner, //
+ behavior.ordinal() >= CoprocessorBehavior.SCAN_FILTER.ordinal(), //
behavior.ordinal() >= CoprocessorBehavior.SCAN_FILTER_AGGR.ordinal());
ByteBuffer buffer = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
@@ -226,7 +238,8 @@ public class CubeVisitService extends CubeVisitProtos.CubeVisitService implement
setFreeSwapSpaceSize(freeSwapSpaceSize).//
setHostname(InetAddress.getLocalHost().getHostName()).//
setEtcMsg(sb.toString()).//
- build()).//
+ build())
+ .//
build());
} catch (IOException ioe) {
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/filter/MassInValueProviderFactoryImpl.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/filter/MassInValueProviderFactoryImpl.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/filter/MassInValueProviderFactoryImpl.java
new file mode 100644
index 0000000..fe6276f
--- /dev/null
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/filter/MassInValueProviderFactoryImpl.java
@@ -0,0 +1,43 @@
+/*
+ * 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.storage.hbase.cube.v2.filter;
+
+import org.apache.kylin.dimension.DimensionEncoding;
+import org.apache.kylin.metadata.filter.UDF.MassInValueProvider;
+import org.apache.kylin.metadata.filter.UDF.MassInValueProviderFactory;
+import org.apache.kylin.metadata.filter.function.Functions;
+import org.apache.kylin.metadata.model.TblColRef;
+
+public class MassInValueProviderFactoryImpl implements MassInValueProviderFactory {
+
+ public interface DimEncAware {
+ DimensionEncoding getDimEnc(TblColRef col);
+ }
+
+ private DimEncAware dimEncAware = null;
+
+ public MassInValueProviderFactoryImpl(DimEncAware dimEncAware) {
+ this.dimEncAware = dimEncAware;
+ }
+
+ @Override
+ public MassInValueProvider getProvider(Functions.FilterTableType filterTableType, String filterResourceIdentifier, TblColRef col) {
+ return new MassInValueProviderImpl(filterTableType, filterResourceIdentifier, dimEncAware.getDimEnc(col));
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/4adea167/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/filter/MassInValueProviderImpl.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/filter/MassInValueProviderImpl.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/filter/MassInValueProviderImpl.java
new file mode 100644
index 0000000..83a6671
--- /dev/null
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/filter/MassInValueProviderImpl.java
@@ -0,0 +1,80 @@
+/*
+ * 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.storage.hbase.cube.v2.filter;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.kylin.common.util.ByteArray;
+import org.apache.kylin.dimension.DimensionEncoding;
+import org.apache.kylin.metadata.filter.UDF.MassInValueProvider;
+import org.apache.kylin.metadata.filter.function.Functions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Stopwatch;
+import com.google.common.collect.Sets;
+
+public class MassInValueProviderImpl implements MassInValueProvider {
+ public static final Logger logger = LoggerFactory.getLogger(MassInValueProviderImpl.class);
+
+ private Set<ByteArray> ret = Sets.newHashSet();
+
+ public MassInValueProviderImpl(Functions.FilterTableType filterTableType, String filterResourceIdentifier, DimensionEncoding encoding) {
+
+ if (filterTableType == Functions.FilterTableType.HDFS) {
+
+ logger.info("Start to load HDFS filter table from " + filterResourceIdentifier);
+ Stopwatch stopwatch = new Stopwatch().start();
+
+ FileSystem fileSystem;
+ try {
+ fileSystem = FileSystem.get(HBaseConfiguration.create());
+ InputStream inputStream = fileSystem.open(new Path(filterResourceIdentifier));
+ List<String> lines = IOUtils.readLines(inputStream);
+
+ logger.info("Load HDFS finished after " + stopwatch.elapsedMillis() + " millis");
+
+ for (String line : lines) {
+ ByteArray byteArray = ByteArray.allocate(encoding.getLengthOfEncoding());
+ encoding.encode(line.getBytes(), line.getBytes().length, byteArray.array(), 0);
+ ret.add(byteArray);
+ }
+
+ logger.info("Mass In values constructed after " + stopwatch.elapsedMillis() + " millis, containing " + ret.size() + " entries");
+
+ } catch (IOException e) {
+ throw new RuntimeException("error when loading the mass in values", e);
+ }
+ } else {
+ throw new RuntimeException("HBASE_TABLE FilterTableType Not supported yet");
+ }
+ }
+
+ @Override
+ public Set<?> getMassInValues() {
+ return ret;
+ }
+}
[2/4] kylin git commit: Support Massin UDF
Posted by ma...@apache.org.
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 <ma...@apache.org>
Authored: Mon Mar 21 13:43:32 2016 +0800
Committer: Hongbin Ma <ma...@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
[4/4] kylin git commit: KYLIN-1526 Move query cache back to query
controller level
Posted by ma...@apache.org.
KYLIN-1526 Move query cache back to query controller level
temp
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/1a13952e
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/1a13952e
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/1a13952e
Branch: refs/heads/master
Commit: 1a13952eadaee154bfb1920653bf741d12c1001a
Parents: 8703220
Author: Hongbin Ma <ma...@apache.org>
Authored: Wed Mar 23 11:15:41 2016 +0800
Committer: Hongbin Ma <ma...@apache.org>
Committed: Thu Mar 24 14:45:02 2016 +0800
----------------------------------------------------------------------
.../org/apache/kylin/common/util/BasicTest.java | 1 -
.../kylin/metadata/tuple/TeeTupleIterator.java | 71 -------
.../metadata/tuple/TeeTupleItrListener.java | 27 ---
.../kylin/storage/ICachableStorageQuery.java | 51 -----
.../cache/AbstractCacheFledgedQuery.java | 100 ----------
.../storage/cache/CacheFledgedDynamicQuery.java | 180 ------------------
.../storage/cache/CacheFledgedStaticQuery.java | 77 --------
.../kylin/storage/cache/StreamSQLResult.java | 103 ----------
.../kylin/storage/cache/DynamicCacheTest.java | 186 -------------------
.../kylin/storage/cache/StaticCacheTest.java | 141 --------------
.../apache/kylin/query/ITKylinQueryTest.java | 11 +-
.../kylin/storage/hbase/ITStorageTest.java | 23 +--
.../kylin/query/enumerator/OLAPEnumerator.java | 1 -
.../kylin/rest/controller/QueryController.java | 37 +++-
.../apache/kylin/rest/response/SQLResponse.java | 3 +
.../apache/kylin/rest/service/CacheService.java | 8 +-
.../apache/kylin/rest/service/QueryService.java | 11 +-
.../org/apache/kylin/rest/StorageCacheTest.java | 116 ------------
.../kylin/storage/hbase/HBaseStorage.java | 34 +---
.../storage/hbase/cube/v1/CubeStorageQuery.java | 24 +--
.../storage/hbase/cube/v2/CubeStorageQuery.java | 22 +--
.../hbase/ii/InvertedIndexStorageQuery.java | 20 +-
.../hbase/cube/CubeStorageQueryTest.java | 8 +-
23 files changed, 66 insertions(+), 1189 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java
----------------------------------------------------------------------
diff --git a/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java b/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java
index 8fcb66a..f1f5aa4 100644
--- a/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java
+++ b/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java
@@ -33,7 +33,6 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import org.apache.kylin.common.KylinVersion;
import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.LoggerFactory;
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/core-metadata/src/main/java/org/apache/kylin/metadata/tuple/TeeTupleIterator.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/tuple/TeeTupleIterator.java b/core-metadata/src/main/java/org/apache/kylin/metadata/tuple/TeeTupleIterator.java
deleted file mode 100644
index e67443f..0000000
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/tuple/TeeTupleIterator.java
+++ /dev/null
@@ -1,71 +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.metadata.tuple;
-
-import java.util.List;
-
-import com.google.common.collect.Lists;
-
-/**
- * Like "tee" command in linux, it effectively duplicates the underlying
- * ITupleIterator's results
- */
-public class TeeTupleIterator implements ITupleIterator {
-
- private ITupleIterator underlying;
- private List<ITuple> duplicatedData;
- private List<TeeTupleItrListener> listeners = Lists.newArrayList();
- private long createTime;
-
- public TeeTupleIterator(ITupleIterator underlying) {
- this.underlying = underlying;
- this.duplicatedData = Lists.newArrayList();
- this.createTime = System.currentTimeMillis();
- }
-
- @Override
- public void close() {
- this.underlying.close();
-
- for (TeeTupleItrListener listener : this.listeners) {
- listener.notify(this.duplicatedData, this.createTime);
- }
- }
-
- @Override
- public boolean hasNext() {
- return this.underlying.hasNext();
- }
-
- @Override
- public ITuple next() {
- ITuple ret = this.underlying.next();
- duplicatedData.add(ret.makeCopy());
- return ret;
- }
-
- @Override
- public void remove() {
- this.underlying.remove();
- }
-
- public void addCloseListener(TeeTupleItrListener listener) {
- this.listeners.add(listener);
- }
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/core-metadata/src/main/java/org/apache/kylin/metadata/tuple/TeeTupleItrListener.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/metadata/tuple/TeeTupleItrListener.java b/core-metadata/src/main/java/org/apache/kylin/metadata/tuple/TeeTupleItrListener.java
deleted file mode 100644
index b5fca16..0000000
--- a/core-metadata/src/main/java/org/apache/kylin/metadata/tuple/TeeTupleItrListener.java
+++ /dev/null
@@ -1,27 +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.metadata.tuple;
-
-import java.util.List;
-
-/**
- */
-public interface TeeTupleItrListener {
- void notify(List<ITuple> duplicated, long createTime);
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/core-storage/src/main/java/org/apache/kylin/storage/ICachableStorageQuery.java
----------------------------------------------------------------------
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/ICachableStorageQuery.java b/core-storage/src/main/java/org/apache/kylin/storage/ICachableStorageQuery.java
deleted file mode 100644
index f401f54..0000000
--- a/core-storage/src/main/java/org/apache/kylin/storage/ICachableStorageQuery.java
+++ /dev/null
@@ -1,51 +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.storage;
-
-import com.google.common.collect.Range;
-
-/**
- */
-public interface ICachableStorageQuery extends IStorageQuery {
- /**
- *
- * being dynamic => getVolatilePeriod() return not null
- * being dynamic => partition column of its realization not null
- *
- * @return true for static storage like cubes
- * false for dynamic storage like II
- */
- boolean isDynamic();
-
- /**
- * volatile period is the period of time in which the returned data is not stable
- * e.g. inverted index's last several minutes' data is dynamic as time goes by.
- * data in this period cannot be cached
- *
- * This method should not be called before ITupleIterator.close() is called
- *
- * @return null if the underlying storage guarantees the data is static
- */
- Range<Long> getVolatilePeriod();
-
- /**
- * get the uuid for the realization that serves this query
- */
- String getStorageUUID();
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/core-storage/src/main/java/org/apache/kylin/storage/cache/AbstractCacheFledgedQuery.java
----------------------------------------------------------------------
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/cache/AbstractCacheFledgedQuery.java b/core-storage/src/main/java/org/apache/kylin/storage/cache/AbstractCacheFledgedQuery.java
deleted file mode 100644
index 4b94cb6..0000000
--- a/core-storage/src/main/java/org/apache/kylin/storage/cache/AbstractCacheFledgedQuery.java
+++ /dev/null
@@ -1,100 +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.storage.cache;
-
-import net.sf.ehcache.Cache;
-import net.sf.ehcache.CacheManager;
-import net.sf.ehcache.Element;
-import net.sf.ehcache.Status;
-import net.sf.ehcache.config.CacheConfiguration;
-
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.metadata.realization.StreamSQLDigest;
-import org.apache.kylin.metadata.tuple.TeeTupleItrListener;
-import org.apache.kylin.storage.ICachableStorageQuery;
-import org.apache.kylin.storage.IStorageQuery;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- */
-public abstract class AbstractCacheFledgedQuery implements IStorageQuery, TeeTupleItrListener {
- private static final Logger logger = LoggerFactory.getLogger(AbstractCacheFledgedQuery.class);
-
- private static final String storageCacheTemplate = "StorageCache";
-
- protected static CacheManager CACHE_MANAGER;
-
- protected ICachableStorageQuery underlyingStorage;
- protected StreamSQLDigest streamSQLDigest;
-
- public AbstractCacheFledgedQuery(ICachableStorageQuery underlyingStorage) {
- this.underlyingStorage = underlyingStorage;
- this.makeCacheIfNecessary(underlyingStorage.getStorageUUID());
- }
-
- public static void setCacheManager(CacheManager cacheManager) {
- CACHE_MANAGER = cacheManager;
- }
-
- protected StreamSQLResult getStreamSQLResult(StreamSQLDigest streamSQLDigest) {
-
- Cache cache = CACHE_MANAGER.getCache(this.underlyingStorage.getStorageUUID());
- Element element = cache.get(streamSQLDigest.hashCode());//TODO: hash code cannot guarantee uniqueness
- if (element != null) {
- return (StreamSQLResult) element.getObjectValue();
- }
- return null;
- }
-
- protected boolean needSaveCache(long createTime) {
- long storageQueryTime = System.currentTimeMillis() - createTime;
- long durationThreshold = KylinConfig.getInstanceFromEnv().getQueryDurationCacheThreshold();
- //TODO: check scan count necessary?
-
- if (storageQueryTime < durationThreshold) {
- logger.info("Skip saving storage caching for storage cache because storage query time {} less than {}", storageQueryTime, durationThreshold);
- return false;
- }
-
- return true;
- }
-
- private void makeCacheIfNecessary(String storageUUID) {
- if (CACHE_MANAGER == null || (!(CACHE_MANAGER.getStatus().equals(Status.STATUS_ALIVE)))) {
- throw new RuntimeException("CACHE_MANAGER is not provided or not alive");
- }
-
- if (CACHE_MANAGER.getCache(storageUUID) == null) {
- logger.info("Cache for {} initializing...", storageUUID);
-
- //Create a Cache specifying its configuration.
- CacheConfiguration templateConf = CACHE_MANAGER.getCache(storageCacheTemplate).getCacheConfiguration();
-
- Cache storageCache = new Cache(new CacheConfiguration(storageUUID, 0).//
- memoryStoreEvictionPolicy(templateConf.getMemoryStoreEvictionPolicy()).//
- eternal(templateConf.isEternal()).//
- timeToIdleSeconds(templateConf.getTimeToIdleSeconds()).//
- //maxBytesLocalHeap(templateConf.getMaxBytesLocalHeap(), MemoryUnit.BYTES).//using pooled size
- persistence(templateConf.getPersistenceConfiguration()));
-
- CACHE_MANAGER.addCacheIfAbsent(storageCache);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/core-storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedDynamicQuery.java
----------------------------------------------------------------------
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedDynamicQuery.java b/core-storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedDynamicQuery.java
deleted file mode 100644
index a93e746..0000000
--- a/core-storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedDynamicQuery.java
+++ /dev/null
@@ -1,180 +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.storage.cache;
-
-import java.util.List;
-
-import net.sf.ehcache.Element;
-
-import org.apache.kylin.common.util.RangeUtil;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.metadata.realization.SQLDigest;
-import org.apache.kylin.metadata.realization.SQLDigestUtil;
-import org.apache.kylin.metadata.realization.StreamSQLDigest;
-import org.apache.kylin.metadata.tuple.CompoundTupleIterator;
-import org.apache.kylin.metadata.tuple.ITuple;
-import org.apache.kylin.metadata.tuple.ITupleIterator;
-import org.apache.kylin.metadata.tuple.SimpleTupleIterator;
-import org.apache.kylin.metadata.tuple.TeeTupleIterator;
-import org.apache.kylin.metadata.tuple.TupleInfo;
-import org.apache.kylin.storage.ICachableStorageQuery;
-import org.apache.kylin.storage.StorageContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Function;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Range;
-
-/**
- */
-public class CacheFledgedDynamicQuery extends AbstractCacheFledgedQuery {
- private static final Logger logger = LoggerFactory.getLogger(CacheFledgedDynamicQuery.class);
-
- private final TblColRef partitionColRef;
-
- private boolean noCacheUsed = true;
-
- private Range<Long> ts;
-
- public CacheFledgedDynamicQuery(ICachableStorageQuery underlyingStorage, TblColRef partitionColRef) {
- super(underlyingStorage);
- this.partitionColRef = partitionColRef;
-
- Preconditions.checkArgument(this.partitionColRef != null, "For dynamic columns like " + //
- this.underlyingStorage.getStorageUUID() + ", partition column must be provided");
- }
-
- @Override
- public ITupleIterator search(final StorageContext context, final SQLDigest sqlDigest, final TupleInfo returnTupleInfo) {
- //check if ts condition in sqlDigest valid
- ts = TsConditionExtractor.extractTsCondition(partitionColRef, sqlDigest.filter);
- if (ts == null || ts.isEmpty()) {
- logger.info("ts range in the query conflicts,return empty directly");
- return ITupleIterator.EMPTY_TUPLE_ITERATOR;
- }
-
- ITupleIterator ret = null;
-
- //enable dynamic cache iff group by columns contains partition col
- //because cache extraction requires partition col value as selection key
- boolean enableDynamicCache = sqlDigest.groupbyColumns.contains(partitionColRef);
-
- if (enableDynamicCache) {
- streamSQLDigest = new StreamSQLDigest(sqlDigest, partitionColRef);
- StreamSQLResult cachedResult = getStreamSQLResult(streamSQLDigest);
- if (cachedResult != null) {
- ret = tryReuseCache(context, sqlDigest, returnTupleInfo, cachedResult);
- } else {
- logger.info("no cache entry for this query");
- }
- }
-
- if (ret == null) {
- ret = underlyingStorage.search(context, sqlDigest, returnTupleInfo);
- logger.info("No Cache being used");
- } else {
- logger.info("Cache being used");
- }
-
- if (enableDynamicCache) {
- //use another nested ITupleIterator to deal with cache
- final TeeTupleIterator tee = new TeeTupleIterator(ret);
- tee.addCloseListener(this);
- return tee;
- } else {
- return ret;
- }
- }
-
- /**
- * if cache is not enough it will try to combine existing cache as well as fresh records
- */
- private ITupleIterator tryReuseCache(final StorageContext context, final SQLDigest sqlDigest, final TupleInfo returnTupleInfo, StreamSQLResult cachedResult) {
- Range<Long> reusePeriod = cachedResult.getReusableResults(ts);
-
- logger.info("existing cache: " + cachedResult);
- logger.info("ts Range in query: " + RangeUtil.formatTsRange(ts));
- logger.info("potential reusable range: " + RangeUtil.formatTsRange(reusePeriod));
-
- if (reusePeriod != null) {
- List<Range<Long>> remainings = RangeUtil.remove(ts, reusePeriod);
- if (remainings.size() == 1) {//if using cache causes two underlyingStorage searches, we'd rather not use the cache
-
- SimpleTupleIterator reusedTuples = new SimpleTupleIterator(cachedResult.reuse(reusePeriod));
- List<ITupleIterator> iTupleIteratorList = Lists.newArrayList();
- iTupleIteratorList.add(reusedTuples);
-
- Range<Long> remaining = remainings.get(0);
- logger.info("Appending ts " + RangeUtil.formatTsRange(remaining) + " as additional filter");
-
- ITupleIterator freshTuples = SQLDigestUtil.appendTsFilterToExecute(sqlDigest, partitionColRef, remaining, new Function<Void, ITupleIterator>() {
- @Override
- public ITupleIterator apply(Void input) {
- return underlyingStorage.search(context, sqlDigest, returnTupleInfo);
- }
- });
- iTupleIteratorList.add(freshTuples);
-
- context.setReusedPeriod(reusePeriod);
- return new CompoundTupleIterator(iTupleIteratorList);
- } else if (remainings.size() == 0) {
- logger.info("The ts range in new query was fully cached");
- context.setReusedPeriod(reusePeriod);
- return new SimpleTupleIterator(cachedResult.reuse(reusePeriod));
- } else {
- //if using cache causes more than one underlyingStorage searches
- //the incurred overhead might be more expensive than the cache benefit
- logger.info("Give up using cache to avoid complexity");
- return null;
- }
- } else {
- logger.info("cached results not reusable by current query");
- return null;
- }
- }
-
- @Override
- public void notify(List<ITuple> duplicated, long createTime) {
-
- //for streaming sql only check if needSaveCache at first entry of cache
- if (noCacheUsed && !needSaveCache(createTime)) {
- return;
- }
-
- Range<Long> cacheExclude = this.underlyingStorage.getVolatilePeriod();
- if (cacheExclude != null) {
- List<Range<Long>> cachablePeriods = RangeUtil.remove(ts, cacheExclude);
- if (cachablePeriods.size() == 1) {
- if (!ts.equals(cachablePeriods.get(0))) {
- logger.info("tsRange shrinks from " + RangeUtil.formatTsRange(ts) + " to " + RangeUtil.formatTsRange(cachablePeriods.get(0)));
- }
- ts = cachablePeriods.get(0);
- } else {
- //give up updating the cache, in avoid to make cache complicated
- logger.info("Skip updating cache to avoid complexity");
- }
- }
-
- StreamSQLResult newCacheEntry = new StreamSQLResult(duplicated, ts, partitionColRef);
- CACHE_MANAGER.getCache(this.underlyingStorage.getStorageUUID()).put(new Element(streamSQLDigest.hashCode(), newCacheEntry));
- logger.info("cache after the query: " + newCacheEntry);
- }
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/core-storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedStaticQuery.java
----------------------------------------------------------------------
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedStaticQuery.java b/core-storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedStaticQuery.java
deleted file mode 100644
index 45c13fe..0000000
--- a/core-storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedStaticQuery.java
+++ /dev/null
@@ -1,77 +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.storage.cache;
-
-import java.util.List;
-
-import net.sf.ehcache.Element;
-
-import org.apache.kylin.metadata.realization.SQLDigest;
-import org.apache.kylin.metadata.realization.StreamSQLDigest;
-import org.apache.kylin.metadata.tuple.ITuple;
-import org.apache.kylin.metadata.tuple.ITupleIterator;
-import org.apache.kylin.metadata.tuple.SimpleTupleIterator;
-import org.apache.kylin.metadata.tuple.TeeTupleIterator;
-import org.apache.kylin.metadata.tuple.TupleInfo;
-import org.apache.kylin.storage.ICachableStorageQuery;
-import org.apache.kylin.storage.StorageContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.collect.Ranges;
-
-public class CacheFledgedStaticQuery extends AbstractCacheFledgedQuery {
- private static final Logger logger = LoggerFactory.getLogger(CacheFledgedStaticQuery.class);
-
- public CacheFledgedStaticQuery(ICachableStorageQuery underlyingStorage) {
- super(underlyingStorage);
- }
-
- @Override
- public ITupleIterator search(final StorageContext context, final SQLDigest sqlDigest, final TupleInfo returnTupleInfo) {
-
- streamSQLDigest = new StreamSQLDigest(sqlDigest, null);
- StreamSQLResult cachedResult = getStreamSQLResult(streamSQLDigest);
- ITupleIterator ret;
-
- if (cachedResult != null) {
- logger.info("using existing cache");
- context.setReusedPeriod(Ranges.<Long> all());
- return new SimpleTupleIterator(cachedResult.reuse(Ranges.<Long> all()));
- } else {
- logger.info("no existing cache to use");
- ret = underlyingStorage.search(context, sqlDigest, returnTupleInfo);
-
- //use another nested ITupleIterator to deal with cache
- final TeeTupleIterator tee = new TeeTupleIterator(ret);
- tee.addCloseListener(this);
- return tee;
- }
- }
-
- @Override
- public void notify(List<ITuple> duplicated, long createTime) {
- if (needSaveCache(createTime)) {
- StreamSQLResult newCacheEntry = new StreamSQLResult(duplicated, Ranges.<Long> all(), null);
- CACHE_MANAGER.getCache(this.underlyingStorage.getStorageUUID()).put(new Element(streamSQLDigest.hashCode(), newCacheEntry));
- logger.info("cache after the query: " + newCacheEntry);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/core-storage/src/main/java/org/apache/kylin/storage/cache/StreamSQLResult.java
----------------------------------------------------------------------
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/cache/StreamSQLResult.java b/core-storage/src/main/java/org/apache/kylin/storage/cache/StreamSQLResult.java
deleted file mode 100644
index f6809df..0000000
--- a/core-storage/src/main/java/org/apache/kylin/storage/cache/StreamSQLResult.java
+++ /dev/null
@@ -1,103 +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.storage.cache;
-
-import java.util.Iterator;
-import java.util.List;
-import java.util.NavigableMap;
-
-import javax.annotation.Nullable;
-
-import org.apache.kylin.common.util.RangeUtil;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.metadata.tuple.ITuple;
-import org.apache.kylin.metadata.tuple.Tuple;
-
-import com.google.common.base.Function;
-import com.google.common.collect.Iterators;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Range;
-
-/**
- */
-public class StreamSQLResult {
- private Range<Long> timeCovered;
- private NavigableMap<Long, List<ITuple>> sortedRows;
-
- public StreamSQLResult(List<ITuple> rows, Range<Long> timeCovered, TblColRef partitionCol) {
-
- sortedRows = Maps.newTreeMap();
- for (ITuple row : rows) {
-
- if (partitionCol != null) {
- long t = Tuple.getTs(row, partitionCol);
-
- //will only cache rows that are within the time range
- if (timeCovered.contains(t)) {
- if (!this.sortedRows.containsKey(t)) {
- this.sortedRows.put(t, Lists.newArrayList(row));
- } else {
- this.sortedRows.get(t).add(row);
- }
- }
- } else {
- if (!this.sortedRows.containsKey(0L)) {
- this.sortedRows.put(0L, Lists.<ITuple> newArrayList());
- }
- this.sortedRows.get(0L).add(row);
- }
- }
- this.timeCovered = timeCovered;
- }
-
- public Range<Long> getReusableResults(Range<Long> tsRange) {
-
- if (tsRange.equals(timeCovered))
- return timeCovered;
-
- if (!timeCovered.isConnected(tsRange)) {
- //share nothing in common
- return null;
- }
-
- Range<Long> ret = timeCovered.intersection(tsRange);
- return ret.isEmpty() ? null : ret;
- }
-
- public Iterator<ITuple> reuse(Range<Long> reusablePeriod) {
- NavigableMap<Long, List<ITuple>> submap = RangeUtil.filter(sortedRows, reusablePeriod);
- return Iterators.concat(Iterators.transform(submap.values().iterator(), new Function<List<ITuple>, Iterator<ITuple>>() {
- @Nullable
- @Override
- public Iterator<ITuple> apply(List<ITuple> input) {
- return input.iterator();
- }
- }));
- }
-
- @Override
- public String toString() {
- return sortedRows.size() + " tuples cached for period " + RangeUtil.formatTsRange(timeCovered);
- }
-
- public long getEndTime() {
- return this.timeCovered.upperEndpoint();
- }
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/core-storage/src/test/java/org/apache/kylin/storage/cache/DynamicCacheTest.java
----------------------------------------------------------------------
diff --git a/core-storage/src/test/java/org/apache/kylin/storage/cache/DynamicCacheTest.java b/core-storage/src/test/java/org/apache/kylin/storage/cache/DynamicCacheTest.java
deleted file mode 100644
index 0adaff3..0000000
--- a/core-storage/src/test/java/org/apache/kylin/storage/cache/DynamicCacheTest.java
+++ /dev/null
@@ -1,186 +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.storage.cache;
-
-import java.util.ArrayList;
-import java.util.IdentityHashMap;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import net.sf.ehcache.CacheManager;
-import org.apache.commons.lang.NotImplementedException;
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.util.DateFormat;
-import org.apache.kylin.common.util.IdentityUtils;
-import org.apache.kylin.metadata.model.FunctionDesc;
-import org.apache.kylin.metadata.model.MeasureDesc;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.metadata.realization.SQLDigest;
-import org.apache.kylin.metadata.tuple.ITuple;
-import org.apache.kylin.metadata.tuple.ITupleIterator;
-import org.apache.kylin.metadata.tuple.SimpleTupleIterator;
-import org.apache.kylin.metadata.tuple.Tuple;
-import org.apache.kylin.metadata.tuple.TupleInfo;
-import org.apache.kylin.storage.ICachableStorageQuery;
-import org.apache.kylin.storage.StorageContext;
-import org.junit.AfterClass;
-import org.junit.Assert;
-import org.junit.BeforeClass;
-import org.junit.Test;
-
-import com.google.common.collect.Lists;
-import com.google.common.collect.Range;
-import com.google.common.collect.Ranges;
-
-/**
- */
-public class DynamicCacheTest {
-
- private static CacheManager cacheManager;
-
- @BeforeClass
- public static void setupResource() throws Exception {
- System.setProperty(KylinConfig.KYLIN_CONF, "../examples/test_case_data/sandbox");
- KylinConfig.getInstanceFromEnv().setProperty("kylin.query.cache.threshold.duration", "0");
-
- cacheManager = CacheManager.newInstance("../server/src/main/resources/ehcache-test.xml");
- AbstractCacheFledgedQuery.setCacheManager(cacheManager);
- }
-
- @AfterClass
- public static void tearDownResource() {
- cacheManager.shutdown();
- AbstractCacheFledgedQuery.setCacheManager(null);
- }
-
-
- class TsOnlyTuple implements ITuple {
- private TblColRef partitionCol;
- private String tsStr;
-
- public TsOnlyTuple(TblColRef partitionCol, String tsStr) {
- this.partitionCol = partitionCol;
- this.tsStr = tsStr;
- }
-
- @Override
- public List<String> getAllFields() {
- throw new NotImplementedException();
- }
-
- @Override
- public List<TblColRef> getAllColumns() {
- throw new NotImplementedException();
- }
-
- @Override
- public Object[] getAllValues() {
- throw new NotImplementedException();
- }
-
- @Override
- public ITuple makeCopy() {
- return new TsOnlyTuple(this.partitionCol, this.tsStr);
- }
-
- @Override
- public Object getValue(TblColRef col) {
- if (col.equals(partitionCol)) {
- return Tuple.dateToEpicDays(this.tsStr);
- } else {
- throw new NotImplementedException();
- }
- }
- }
-
- @Test
- public void basicTest() {
-
- final StorageContext context = new StorageContext();
- final List<TblColRef> groups = StorageMockUtils.buildGroups();
- final TblColRef partitionCol = groups.get(0);
- final List<FunctionDesc> aggregations = StorageMockUtils.buildAggregations();
- final TupleInfo tupleInfo = StorageMockUtils.newTupleInfo(groups, aggregations);
-
- SQLDigest sqlDigest = new SQLDigest("default.test_kylin_fact", null, null, Lists.<TblColRef> newArrayList(), groups, Lists.newArrayList(partitionCol), Lists.<TblColRef> newArrayList(), aggregations, new ArrayList<MeasureDesc>(), new ArrayList<SQLDigest.OrderEnum>());
-
- ITuple aTuple = new TsOnlyTuple(partitionCol, "2011-02-01");
- ITuple bTuple = new TsOnlyTuple(partitionCol, "2012-02-01");
- final List<ITuple> allTuples = Lists.newArrayList(aTuple, bTuple);
-
- //counts for verifying
- final AtomicInteger underlyingSEHitCount = new AtomicInteger(0);
- final List<Integer> returnedRowPerSearch = Lists.newArrayList();
-
- CacheFledgedDynamicQuery dynamicCache = new CacheFledgedDynamicQuery(new ICachableStorageQuery() {
- @Override
- public ITupleIterator search(StorageContext context, SQLDigest sqlDigest, TupleInfo returnTupleInfo) {
- Range<Long> tsRagneInQuery = TsConditionExtractor.extractTsCondition(partitionCol, sqlDigest.filter);
- List<ITuple> ret = Lists.newArrayList();
- for (ITuple tuple : allTuples) {
- if (tsRagneInQuery.contains(Tuple.getTs(tuple, partitionCol))) {
- ret.add(tuple);
- }
- }
-
- underlyingSEHitCount.incrementAndGet();
- returnedRowPerSearch.add(ret.size());
-
- return new SimpleTupleIterator(ret.iterator());
- }
-
- @Override
- public boolean isDynamic() {
- return true;
- }
-
- @Override
- public Range<Long> getVolatilePeriod() {
- return Ranges.greaterThan(DateFormat.stringToMillis("2011-02-01"));
- }
-
- @Override
- public String getStorageUUID() {
- return "111ca32a-a33e-4b69-12aa-0bb8b1f8c191";
- }
- }, partitionCol);
-
- sqlDigest.filter = StorageMockUtils.buildTs2010Filter(groups.get(0));
- ITupleIterator firstIterator = dynamicCache.search(context, sqlDigest, tupleInfo);
- IdentityHashMap<ITuple, Void> firstResults = new IdentityHashMap<>();
- while (firstIterator.hasNext()) {
- firstResults.put(firstIterator.next(), null);
- }
- firstIterator.close();
-
- sqlDigest.filter = StorageMockUtils.buildTs2011Filter(groups.get(0));
- ITupleIterator secondIterator = dynamicCache.search(context, sqlDigest, tupleInfo);
- IdentityHashMap<ITuple, Void> secondResults = new IdentityHashMap<>();
- while (secondIterator.hasNext()) {
- secondResults.put(secondIterator.next(), null);
- }
- secondIterator.close();
-
- Assert.assertEquals(2, firstResults.size());
- IdentityUtils.collectionReferenceEquals(firstResults.keySet(), secondResults.keySet());
- Assert.assertEquals(2, underlyingSEHitCount.get());
- Assert.assertEquals(new Integer(2), returnedRowPerSearch.get(0));
- Assert.assertEquals(new Integer(1), returnedRowPerSearch.get(1));
- }
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/core-storage/src/test/java/org/apache/kylin/storage/cache/StaticCacheTest.java
----------------------------------------------------------------------
diff --git a/core-storage/src/test/java/org/apache/kylin/storage/cache/StaticCacheTest.java b/core-storage/src/test/java/org/apache/kylin/storage/cache/StaticCacheTest.java
deleted file mode 100644
index 0a4795f..0000000
--- a/core-storage/src/test/java/org/apache/kylin/storage/cache/StaticCacheTest.java
+++ /dev/null
@@ -1,141 +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.storage.cache;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.IdentityHashMap;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import net.sf.ehcache.CacheManager;
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.util.IdentityUtils;
-import org.apache.kylin.metadata.filter.TupleFilter;
-import org.apache.kylin.metadata.model.FunctionDesc;
-import org.apache.kylin.metadata.model.MeasureDesc;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.metadata.realization.SQLDigest;
-import org.apache.kylin.metadata.tuple.ITuple;
-import org.apache.kylin.metadata.tuple.ITupleIterator;
-import org.apache.kylin.metadata.tuple.SimpleTupleIterator;
-import org.apache.kylin.metadata.tuple.Tuple;
-import org.apache.kylin.metadata.tuple.TupleInfo;
-import org.apache.kylin.storage.ICachableStorageQuery;
-import org.apache.kylin.storage.StorageContext;
-import org.junit.AfterClass;
-import org.junit.Assert;
-import org.junit.BeforeClass;
-import org.junit.Test;
-
-import com.google.common.collect.Lists;
-import com.google.common.collect.Range;
-
-/**
- */
-public class StaticCacheTest {
-
- private static CacheManager cacheManager;
-
- @BeforeClass
- public static void setupResource() throws Exception {
-
- System.setProperty(KylinConfig.KYLIN_CONF, "../examples/test_case_data/sandbox");
-
- KylinConfig config = KylinConfig.getInstanceFromEnv();
- config.setProperty("kylin.query.cache.threshold.duration", "0");
-
- cacheManager = CacheManager.newInstance("../server/src/main/resources/ehcache-test.xml");
- AbstractCacheFledgedQuery.setCacheManager(cacheManager);
- }
-
- @AfterClass
- public static void tearDownResource() {
- cacheManager.shutdown();
- AbstractCacheFledgedQuery.setCacheManager(null);
- }
-
- @Test
- public void basicTest() {
-
- final StorageContext context = new StorageContext();
- final List<TblColRef> groups = StorageMockUtils.buildGroups();
- final List<FunctionDesc> aggregations = StorageMockUtils.buildAggregations();
- final TupleFilter filter = StorageMockUtils.buildFilter1(groups.get(0));
- final SQLDigest sqlDigest = new SQLDigest("default.test_kylin_fact", filter, null, Collections.<TblColRef> emptySet(), groups, Collections.<TblColRef> emptySet(), Collections.<TblColRef> emptySet(), aggregations, new ArrayList<MeasureDesc>(), new ArrayList<SQLDigest.OrderEnum>());
- final TupleInfo tupleInfo = StorageMockUtils.newTupleInfo(groups, aggregations);
-
- final List<ITuple> ret = Lists.newArrayList();
- ret.add(new Tuple(tupleInfo));
- ret.add(new Tuple(tupleInfo));
- ret.add(new Tuple(tupleInfo));
-
- final AtomicInteger underlyingSEHitCount = new AtomicInteger(0);
-
- CacheFledgedStaticQuery cacheFledgedStaticStorageEngine = new CacheFledgedStaticQuery(new ICachableStorageQuery() {
- @Override
- public ITupleIterator search(StorageContext context, SQLDigest sqlDigest, TupleInfo returnTupleInfo) {
- underlyingSEHitCount.incrementAndGet();
- return new SimpleTupleIterator(ret.iterator());
- }
-
- @Override
- public boolean isDynamic() {
- return false;
- }
-
- @Override
- public Range<Long> getVolatilePeriod() {
- return null;
- }
-
- @Override
- public String getStorageUUID() {
- return "111ca32a-a33e-4b69-12aa-0bb8b1f8c092";
- }
- });
-
- ITupleIterator firstIterator = cacheFledgedStaticStorageEngine.search(context, sqlDigest, tupleInfo);
- IdentityHashMap<ITuple, Void> firstResults = new IdentityHashMap<>();
- while (firstIterator.hasNext()) {
- firstResults.put(firstIterator.next(), null);
- }
- firstIterator.close();
-
- ITupleIterator secondIterator = cacheFledgedStaticStorageEngine.search(context, sqlDigest, tupleInfo);
- IdentityHashMap<ITuple, Void> secondResults = new IdentityHashMap<>();
- while (secondIterator.hasNext()) {
- secondResults.put(secondIterator.next(), null);
- }
- secondIterator.close();
-
- ITupleIterator thirdIterator = cacheFledgedStaticStorageEngine.search(context, sqlDigest, tupleInfo);
- IdentityHashMap<ITuple, Void> thirdResults = new IdentityHashMap<>();
- while (thirdIterator.hasNext()) {
- thirdResults.put(thirdIterator.next(), null);
- }
- thirdIterator.close();
-
- Assert.assertEquals(3, firstResults.size());
- IdentityUtils.collectionReferenceEquals(firstResults.keySet(), secondResults.keySet());
- IdentityUtils.collectionReferenceEquals(thirdResults.keySet(), secondResults.keySet());
-
- Assert.assertEquals(1, underlyingSEHitCount.get());
- }
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/kylin-it/src/test/java/org/apache/kylin/query/ITKylinQueryTest.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/query/ITKylinQueryTest.java b/kylin-it/src/test/java/org/apache/kylin/query/ITKylinQueryTest.java
index 3fb93d7..4168df2 100644
--- a/kylin-it/src/test/java/org/apache/kylin/query/ITKylinQueryTest.java
+++ b/kylin-it/src/test/java/org/apache/kylin/query/ITKylinQueryTest.java
@@ -25,16 +25,14 @@ import java.sql.DriverManager;
import java.util.List;
import java.util.Properties;
-import net.sf.ehcache.CacheManager;
import org.apache.commons.lang3.StringUtils;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.HBaseMetadataTestCase;
import org.apache.kylin.metadata.project.ProjectInstance;
import org.apache.kylin.query.enumerator.OLAPQuery;
import org.apache.kylin.query.relnode.OLAPContext;
import org.apache.kylin.query.schema.OLAPSchemaFactory;
-import org.apache.kylin.storage.cache.AbstractCacheFledgedQuery;
import org.apache.kylin.storage.hbase.cube.v1.coprocessor.observer.ObserverEnabler;
-import org.apache.kylin.common.util.HBaseMetadataTestCase;
import org.dbunit.database.DatabaseConnection;
import org.dbunit.database.IDatabaseConnection;
import org.junit.AfterClass;
@@ -44,7 +42,6 @@ import org.junit.Test;
@Ignore("KylinQueryTest is contained by ITCombinationTest")
public class ITKylinQueryTest extends KylinTestBase {
- private static CacheManager cacheManager;
@BeforeClass
public static void setUp() throws Exception {
@@ -77,8 +74,6 @@ public class ITKylinQueryTest extends KylinTestBase {
H2Database h2DB = new H2Database(h2Connection, config);
h2DB.loadAllTables();
- cacheManager = CacheManager.newInstance("../server/src/main/resources/ehcache-test.xml");
- AbstractCacheFledgedQuery.setCacheManager(cacheManager);
}
protected static void clean() {
@@ -90,10 +85,6 @@ public class ITKylinQueryTest extends KylinTestBase {
ObserverEnabler.forceCoprocessorUnset();
HBaseMetadataTestCase.staticCleanupTestMetadata();
- if (cacheManager != null) {
- cacheManager.shutdown();
- }
- AbstractCacheFledgedQuery.setCacheManager(null);
}
@Ignore("this is only for debug")
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITStorageTest.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITStorageTest.java b/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITStorageTest.java
index c253770..15e435e 100644
--- a/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITStorageTest.java
+++ b/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITStorageTest.java
@@ -24,8 +24,6 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import net.sf.ehcache.CacheManager;
-
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.util.HBaseMetadataTestCase;
import org.apache.kylin.cube.CubeInstance;
@@ -40,7 +38,6 @@ import org.apache.kylin.metadata.tuple.ITupleIterator;
import org.apache.kylin.storage.IStorageQuery;
import org.apache.kylin.storage.StorageContext;
import org.apache.kylin.storage.StorageFactory;
-import org.apache.kylin.storage.cache.AbstractCacheFledgedQuery;
import org.apache.kylin.storage.cache.StorageMockUtils;
import org.apache.kylin.storage.exception.ScanOutOfLimitException;
import org.junit.After;
@@ -57,18 +54,12 @@ public class ITStorageTest extends HBaseMetadataTestCase {
private CubeInstance cube;
private StorageContext context;
- private static CacheManager cacheManager;
-
@BeforeClass
public static void setupResource() throws Exception {
- cacheManager = CacheManager.newInstance("../server/src/main/resources/ehcache-test.xml");
- AbstractCacheFledgedQuery.setCacheManager(cacheManager);
}
@AfterClass
public static void tearDownResource() {
- cacheManager.shutdown();
- AbstractCacheFledgedQuery.setCacheManager(null);
}
@Before
@@ -115,36 +106,36 @@ public class ITStorageTest extends HBaseMetadataTestCase {
List<TblColRef> groups = buildGroups();
List<FunctionDesc> aggregations = buildAggregations();
TupleFilter filter = buildFilter2(groups.get(1));
-
+
int count = search(groups, aggregations, filter, context);
assertTrue(count > 0);
}
-
+
@Test
public void test03() {
List<TblColRef> groups = buildGroups();
List<FunctionDesc> aggregations = buildAggregations();
TupleFilter filter = buildAndFilter(groups);
-
+
int count = search(groups, aggregations, filter, context);
assertTrue(count > 0);
}
-
+
@Test
public void test04() {
List<TblColRef> groups = buildGroups();
List<FunctionDesc> aggregations = buildAggregations();
TupleFilter filter = buildOrFilter(groups);
-
+
int count = search(groups, aggregations, filter, context);
assertTrue(count > 0);
}
-
+
@Test
public void test05() {
List<TblColRef> groups = buildGroups();
List<FunctionDesc> aggregations = buildAggregations();
-
+
int count = search(groups, aggregations, null, context);
assertTrue(count > 0);
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/query/src/main/java/org/apache/kylin/query/enumerator/OLAPEnumerator.java
----------------------------------------------------------------------
diff --git a/query/src/main/java/org/apache/kylin/query/enumerator/OLAPEnumerator.java b/query/src/main/java/org/apache/kylin/query/enumerator/OLAPEnumerator.java
index aa8b2a7..07f72b1 100644
--- a/query/src/main/java/org/apache/kylin/query/enumerator/OLAPEnumerator.java
+++ b/query/src/main/java/org/apache/kylin/query/enumerator/OLAPEnumerator.java
@@ -125,7 +125,6 @@ public class OLAPEnumerator implements Enumerator<Object[]> {
ITupleIterator iterator = storageEngine.search(olapContext.storageContext, sqlDigest, olapContext.returnTupleInfo);
if (logger.isDebugEnabled()) {
logger.debug("return TupleIterator...");
- logger.debug("Storage cache used for this storage query:" + olapContext.storageContext.getReusedPeriod());
}
return iterator;
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/server/src/main/java/org/apache/kylin/rest/controller/QueryController.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/kylin/rest/controller/QueryController.java b/server/src/main/java/org/apache/kylin/rest/controller/QueryController.java
index f60894e..88cf912 100644
--- a/server/src/main/java/org/apache/kylin/rest/controller/QueryController.java
+++ b/server/src/main/java/org/apache/kylin/rest/controller/QueryController.java
@@ -46,7 +46,6 @@ import org.apache.kylin.rest.request.SaveSqlRequest;
import org.apache.kylin.rest.response.SQLResponse;
import org.apache.kylin.rest.service.QueryService;
import org.apache.kylin.rest.util.QueryUtil;
-import org.apache.kylin.storage.cache.AbstractCacheFledgedQuery;
import org.apache.kylin.storage.exception.ScanOutOfLimitException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -75,6 +74,7 @@ public class QueryController extends BasicController {
private static final Logger logger = LoggerFactory.getLogger(QueryController.class);
+ public static final String SUCCESS_QUERY_CACHE = "StorageCache";
public static final String EXCEPTION_QUERY_CACHE = "ExceptionQueryCache";
@Autowired
@@ -86,7 +86,6 @@ public class QueryController extends BasicController {
@PostConstruct
public void init() throws IOException {
Preconditions.checkNotNull(cacheManager, "cacheManager is not injected yet");
- AbstractCacheFledgedQuery.setCacheManager(cacheManager);
}
@RequestMapping(value = "/query", method = RequestMethod.POST)
@@ -185,10 +184,23 @@ public class QueryController extends BasicController {
throw new InternalErrorException("Not Supported SQL.");
}
+ long startTime = System.currentTimeMillis();
+
SQLResponse sqlResponse = searchQueryInCache(sqlRequest);
try {
if (null == sqlResponse) {
sqlResponse = queryService.query(sqlRequest);
+
+ long durationThreshold = KylinConfig.getInstanceFromEnv().getQueryDurationCacheThreshold();
+ long scancountThreshold = KylinConfig.getInstanceFromEnv().getQueryScanCountCacheThreshold();
+ sqlResponse.setDuration(System.currentTimeMillis() - startTime);
+ logger.info("Stats of SQL response: isException: {}, duration: {}, total scan count {}", //
+ new String[] { String.valueOf(sqlResponse.getIsException()), String.valueOf(sqlResponse.getDuration()), String.valueOf(sqlResponse.getTotalScanCount()) });
+ if (!sqlResponse.getIsException() && (sqlResponse.getDuration() > durationThreshold || sqlResponse.getTotalScanCount() > scancountThreshold)) {
+ cacheManager.getCache(SUCCESS_QUERY_CACHE).put(new Element(sqlRequest, sqlResponse));
+ }
+ } else {
+ sqlResponse.setDuration(System.currentTimeMillis() - startTime);
}
checkQueryAuth(sqlResponse);
@@ -221,13 +233,20 @@ public class QueryController extends BasicController {
private SQLResponse searchQueryInCache(SQLRequest sqlRequest) {
SQLResponse response = null;
Cache exceptionCache = cacheManager.getCache(EXCEPTION_QUERY_CACHE);
-
- if (KylinConfig.getInstanceFromEnv().isQueryCacheEnabled() && //
- !BackdoorToggles.getDisableCache() && //
- exceptionCache.get(sqlRequest) != null) {
- Element element = exceptionCache.get(sqlRequest);
- response = (SQLResponse) element.getObjectValue();
- response.setHitExceptionCache(true);
+ Cache successCache = cacheManager.getCache(SUCCESS_QUERY_CACHE);
+
+ if (KylinConfig.getInstanceFromEnv().isQueryCacheEnabled() && !BackdoorToggles.getDisableCache()) {
+ if (exceptionCache.get(sqlRequest) != null) {
+ logger.info("The sqlResponse is found in EXCEPTION_QUERY_CACHE");
+ Element element = exceptionCache.get(sqlRequest);
+ response = (SQLResponse) element.getObjectValue();
+ response.setHitExceptionCache(true);
+ } else if (successCache.get(sqlRequest) != null) {
+ logger.info("The sqlResponse is found in SUCCESS_QUERY_CACHE");
+ Element element = successCache.get(sqlRequest);
+ response = (SQLResponse) element.getObjectValue();
+ response.setStorageCacheUsed(true);
+ }
}
return response;
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/server/src/main/java/org/apache/kylin/rest/response/SQLResponse.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/kylin/rest/response/SQLResponse.java b/server/src/main/java/org/apache/kylin/rest/response/SQLResponse.java
index f5ef8c5..8fb8788 100644
--- a/server/src/main/java/org/apache/kylin/rest/response/SQLResponse.java
+++ b/server/src/main/java/org/apache/kylin/rest/response/SQLResponse.java
@@ -35,6 +35,9 @@ public class SQLResponse implements Serializable {
// the results rows, each row contains several columns
private List<List<String>> results;
+ /**
+ * for historical reasons it is named "cube", however it might also refer to any realizations like hybrid, II or etc.
+ */
private String cube;
// if not select query, only return affected row count
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/server/src/main/java/org/apache/kylin/rest/service/CacheService.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/kylin/rest/service/CacheService.java b/server/src/main/java/org/apache/kylin/rest/service/CacheService.java
index 83b9ac7..9a6036d 100644
--- a/server/src/main/java/org/apache/kylin/rest/service/CacheService.java
+++ b/server/src/main/java/org/apache/kylin/rest/service/CacheService.java
@@ -49,6 +49,7 @@ import org.apache.kylin.metadata.realization.RealizationRegistry;
import org.apache.kylin.metadata.realization.RealizationType;
import org.apache.kylin.query.enumerator.OLAPQuery;
import org.apache.kylin.query.schema.OLAPSchemaFactory;
+import org.apache.kylin.rest.controller.QueryController;
import org.apache.kylin.source.kafka.KafkaConfigManager;
import org.apache.kylin.storage.hbase.HBaseConnection;
import org.apache.kylin.storage.hybrid.HybridManager;
@@ -101,9 +102,10 @@ public class CacheService extends BasicService {
}
protected void cleanDataCache(String storageUUID) {
- if (cacheManager != null && cacheManager.getCache(storageUUID) != null) {
- logger.info("cleaning cache for " + storageUUID);
- cacheManager.getCache(storageUUID).removeAll();
+ if (cacheManager != null) {
+ logger.info("cleaning cache for " + storageUUID + " (currently remove all entries)");
+ cacheManager.getCache(QueryController.SUCCESS_QUERY_CACHE).removeAll();
+ cacheManager.getCache(QueryController.EXCEPTION_QUERY_CACHE).removeAll();
} else {
logger.warn("skip cleaning cache for " + storageUUID);
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/server/src/main/java/org/apache/kylin/rest/service/QueryService.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/kylin/rest/service/QueryService.java b/server/src/main/java/org/apache/kylin/rest/service/QueryService.java
index bf371be..059a095 100644
--- a/server/src/main/java/org/apache/kylin/rest/service/QueryService.java
+++ b/server/src/main/java/org/apache/kylin/rest/service/QueryService.java
@@ -203,7 +203,7 @@ public class QueryService extends BasicService {
final Set<String> realizationNames = new HashSet<String>();
final Set<Long> cuboidIds = new HashSet<Long>();
float duration = response.getDuration() / (float) 1000;
- boolean storageCacheUsed = false;
+ boolean storageCacheUsed = response.isStorageCacheUsed();
if (!response.isHitExceptionCache() && null != OLAPContext.getThreadLocalContexts()) {
for (OLAPContext ctx : OLAPContext.getThreadLocalContexts()) {
@@ -218,10 +218,6 @@ public class QueryService extends BasicService {
realizationNames.add(realizationName);
}
- if (ctx.storageContext.getReusedPeriod() != null) {
- response.setStorageCacheUsed(true);
- storageCacheUsed = true;
- }
}
}
@@ -340,7 +336,6 @@ public class QueryService extends BasicService {
Connection conn = null;
Statement stat = null;
ResultSet resultSet = null;
- long startTime = System.currentTimeMillis();
List<List<String>> results = new LinkedList<List<String>>();
List<SelectedColumnMeta> columnMetas = new LinkedList<SelectedColumnMeta>();
@@ -386,6 +381,7 @@ public class QueryService extends BasicService {
boolean isPartialResult = false;
String cube = "";
+ StringBuilder sb = new StringBuilder("Scan count for each storageContext: ");
long totalScanCount = 0;
if (OLAPContext.getThreadLocalContexts() != null) { // contexts can be null in case of 'explain plan for'
for (OLAPContext ctx : OLAPContext.getThreadLocalContexts()) {
@@ -393,13 +389,14 @@ public class QueryService extends BasicService {
isPartialResult |= ctx.storageContext.isPartialResultReturned();
cube = ctx.realization.getName();
totalScanCount += ctx.storageContext.getTotalScanCount();
+ sb.append(ctx.storageContext.getTotalScanCount() + ",");
}
}
}
+ logger.info(sb.toString());
SQLResponse response = new SQLResponse(columnMetas, results, cube, 0, false, null, isPartialResult);
response.setTotalScanCount(totalScanCount);
- response.setDuration(System.currentTimeMillis() - startTime);
return response;
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/server/src/test/java/org/apache/kylin/rest/StorageCacheTest.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/org/apache/kylin/rest/StorageCacheTest.java b/server/src/test/java/org/apache/kylin/rest/StorageCacheTest.java
deleted file mode 100644
index b6cc6d0..0000000
--- a/server/src/test/java/org/apache/kylin/rest/StorageCacheTest.java
+++ /dev/null
@@ -1,116 +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.rest;
-
-import net.sf.ehcache.CacheManager;
-import net.sf.ehcache.Element;
-
-import org.apache.commons.lang.NotImplementedException;
-import org.apache.kylin.metadata.realization.SQLDigest;
-import org.apache.kylin.metadata.tuple.ITupleIterator;
-import org.apache.kylin.metadata.tuple.TupleInfo;
-import org.apache.kylin.rest.service.ServiceTestBase;
-import org.apache.kylin.storage.ICachableStorageQuery;
-import org.apache.kylin.storage.StorageContext;
-import org.apache.kylin.storage.cache.AbstractCacheFledgedQuery;
-import org.apache.kylin.storage.cache.CacheFledgedStaticQuery;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-import org.springframework.beans.factory.annotation.Autowired;
-
-import com.google.common.collect.Range;
-
-public class StorageCacheTest extends ServiceTestBase {
-
- public class MockedCachableStorageQuery implements ICachableStorageQuery {
- private String uuid;
-
- public MockedCachableStorageQuery(String uuid) {
- this.uuid = uuid;
- }
-
- @Override
- public boolean isDynamic() {
- return false;
- }
-
- @Override
- public Range<Long> getVolatilePeriod() {
- throw new NotImplementedException();
- }
-
- @Override
- public String getStorageUUID() {
- return this.uuid;
- }
-
- @Override
- public ITupleIterator search(StorageContext context, SQLDigest sqlDigest, TupleInfo returnTupleInfo) {
- throw new NotImplementedException();
- }
- }
-
- @Autowired
- private CacheManager cacheManager;
-
- @Before
- public void setup() throws Exception {
- AbstractCacheFledgedQuery.setCacheManager(cacheManager);
- new CacheFledgedStaticQuery(new MockedCachableStorageQuery("1"));
- new CacheFledgedStaticQuery(new MockedCachableStorageQuery("2"));
- new CacheFledgedStaticQuery(new MockedCachableStorageQuery("3"));
- new CacheFledgedStaticQuery(new MockedCachableStorageQuery("4"));
- new CacheFledgedStaticQuery(new MockedCachableStorageQuery("5"));
- new CacheFledgedStaticQuery(new MockedCachableStorageQuery("6"));
- new CacheFledgedStaticQuery(new MockedCachableStorageQuery("7"));
- new CacheFledgedStaticQuery(new MockedCachableStorageQuery("8"));
- new CacheFledgedStaticQuery(new MockedCachableStorageQuery("9"));
- new CacheFledgedStaticQuery(new MockedCachableStorageQuery("10"));
- new CacheFledgedStaticQuery(new MockedCachableStorageQuery("11"));
- }
-
- @Test
- public void test1() {
- int oneM = 1 << 20;
- cacheManager.getCache("1").put(new Element("xx", new byte[oneM]));
- Element xx = cacheManager.getCache("1").get("xx");
- Assert.assertEquals(oneM, ((byte[]) xx.getObjectValue()).length);
-
- cacheManager.getCache("2").put(new Element("yy", new byte[3 * oneM]));
- Element yy = cacheManager.getCache("2").get("yy");
- Assert.assertEquals(3 * oneM, ((byte[]) yy.getObjectValue()).length);
-
- cacheManager.getCache("3").put(new Element("zz", new byte[10 * oneM]));
- Element zz = cacheManager.getCache("3").get("zz");
- Assert.assertEquals(null, zz);
-
- cacheManager.getCache("4").put(new Element("aa", new byte[oneM]));
- Element aa = cacheManager.getCache("4").get("aa");
- Assert.assertEquals(oneM, ((byte[]) aa.getObjectValue()).length);
-
- cacheManager.getCache("2").put(new Element("bb", new byte[3 * oneM]));
- Element bb = cacheManager.getCache("2").get("bb");
- Assert.assertEquals(3 * oneM, ((byte[]) bb.getObjectValue()).length);
- }
-
- @Test
- public void test2() {
- }
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseStorage.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseStorage.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseStorage.java
index e71434f..20bc229 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseStorage.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseStorage.java
@@ -31,11 +31,8 @@ import org.apache.kylin.metadata.model.PartitionDesc;
import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.metadata.realization.IRealization;
import org.apache.kylin.metadata.realization.RealizationType;
-import org.apache.kylin.storage.ICachableStorageQuery;
import org.apache.kylin.storage.IStorage;
import org.apache.kylin.storage.IStorageQuery;
-import org.apache.kylin.storage.cache.CacheFledgedDynamicQuery;
-import org.apache.kylin.storage.cache.CacheFledgedStaticQuery;
import org.apache.kylin.storage.hbase.steps.HBaseMROutput;
import org.apache.kylin.storage.hbase.steps.HBaseMROutput2Transition;
@@ -54,22 +51,15 @@ public class HBaseStorage implements IStorage {
@Override
public IStorageQuery createQuery(IRealization realization) {
- boolean queryCacheGloballyEnabled = KylinConfig.getInstanceFromEnv().isQueryCacheEnabled();
- boolean queryCacheQueryLevelEnabled = !BackdoorToggles.getDisableCache();
-
if (realization.getType() == RealizationType.INVERTED_INDEX) {
- ICachableStorageQuery ret;
+ IStorageQuery ret;
try {
- ret = (ICachableStorageQuery) Class.forName(defaultIIStorageQuery).getConstructor(IIInstance.class).newInstance((IIInstance) realization);
+ ret = (IStorageQuery) Class.forName(defaultIIStorageQuery).getConstructor(IIInstance.class).newInstance((IIInstance) realization);
} catch (Exception e) {
throw new RuntimeException("Failed to initialize storage query for " + defaultIIStorageQuery, e);
}
+ return ret;
- if (queryCacheGloballyEnabled && queryCacheQueryLevelEnabled) {
- return wrapWithCache(ret, realization);
- } else {
- return ret;
- }
} else if (realization.getType() == RealizationType.CUBE) {
CubeInstance cubeInstance = (CubeInstance) realization;
@@ -84,31 +74,19 @@ public class HBaseStorage implements IStorage {
cubeStorageQuery = v2CubeStorageQuery;//by default use v2
}
- ICachableStorageQuery ret;
+ IStorageQuery ret;
try {
- ret = (ICachableStorageQuery) Class.forName(cubeStorageQuery).getConstructor(CubeInstance.class).newInstance((CubeInstance) realization);
+ ret = (IStorageQuery) Class.forName(cubeStorageQuery).getConstructor(CubeInstance.class).newInstance((CubeInstance) realization);
} catch (Exception e) {
throw new RuntimeException("Failed to initialize storage query for " + cubeStorageQuery, e);
}
- if (queryCacheGloballyEnabled && queryCacheQueryLevelEnabled) {
- return wrapWithCache(ret, realization);
- } else {
- return ret;
- }
+ return ret;
} else {
throw new IllegalArgumentException("Unknown realization type " + realization.getType());
}
}
- private static IStorageQuery wrapWithCache(ICachableStorageQuery underlyingStorageEngine, IRealization realization) {
- if (underlyingStorageEngine.isDynamic()) {
- return new CacheFledgedDynamicQuery(underlyingStorageEngine, getPartitionCol(realization));
- } else {
- return new CacheFledgedStaticQuery(underlyingStorageEngine);
- }
- }
-
private static TblColRef getPartitionCol(IRealization realization) {
String modelName = realization.getModelName();
DataModelDesc dataModelDesc = MetadataManager.getInstance(KylinConfig.getInstanceFromEnv()).getDataModelDesc(modelName);
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeStorageQuery.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeStorageQuery.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeStorageQuery.java
index 075328f..47e94de 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeStorageQuery.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/CubeStorageQuery.java
@@ -62,7 +62,7 @@ import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.metadata.realization.SQLDigest;
import org.apache.kylin.metadata.tuple.ITupleIterator;
import org.apache.kylin.metadata.tuple.TupleInfo;
-import org.apache.kylin.storage.ICachableStorageQuery;
+import org.apache.kylin.storage.IStorageQuery;
import org.apache.kylin.storage.StorageContext;
import org.apache.kylin.storage.hbase.HBaseConnection;
import org.apache.kylin.storage.hbase.cube.v1.coprocessor.observer.ObserverEnabler;
@@ -75,11 +75,10 @@ import org.slf4j.LoggerFactory;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
-import com.google.common.collect.Range;
import com.google.common.collect.Sets;
@SuppressWarnings("unused")
-public class CubeStorageQuery implements ICachableStorageQuery {
+public class CubeStorageQuery implements IStorageQuery {
private static final Logger logger = LoggerFactory.getLogger(CubeStorageQuery.class);
@@ -154,21 +153,6 @@ public class CubeStorageQuery implements ICachableStorageQuery {
return new SerializedHBaseTupleIterator(conn, scans, cubeInstance, dimensionsD, filterD, groupsCopD, valueDecoders, context, returnTupleInfo);
}
- @Override
- public Range<Long> getVolatilePeriod() {
- return null;
- }
-
- @Override
- public String getStorageUUID() {
- return this.uuid;
- }
-
- @Override
- public boolean isDynamic() {
- return false;
- }
-
private void buildDimensionsAndMetrics(Collection<TblColRef> dimensions, Collection<FunctionDesc> metrics, SQLDigest sqlDigest) {
for (FunctionDesc func : sqlDigest.aggregations) {
@@ -493,7 +477,7 @@ public class CubeStorageQuery implements ICachableStorageQuery {
List<Collection<ColumnValueRange>> result = Lists.newArrayList();
if (flatFilter == null) {
- result.add(Collections.<ColumnValueRange>emptyList());
+ result.add(Collections.<ColumnValueRange> emptyList());
return result;
}
@@ -535,7 +519,7 @@ public class CubeStorageQuery implements ICachableStorageQuery {
}
if (globalAlwaysTrue) {
orAndRanges.clear();
- orAndRanges.add(Collections.<ColumnValueRange>emptyList());
+ orAndRanges.add(Collections.<ColumnValueRange> emptyList());
}
return orAndRanges;
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeStorageQuery.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeStorageQuery.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeStorageQuery.java
index 85082ca..863dd67 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeStorageQuery.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeStorageQuery.java
@@ -45,17 +45,16 @@ import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.metadata.realization.SQLDigest;
import org.apache.kylin.metadata.tuple.ITupleIterator;
import org.apache.kylin.metadata.tuple.TupleInfo;
-import org.apache.kylin.storage.ICachableStorageQuery;
+import org.apache.kylin.storage.IStorageQuery;
import org.apache.kylin.storage.StorageContext;
import org.apache.kylin.storage.translate.DerivedFilterTranslator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.collect.Lists;
-import com.google.common.collect.Range;
import com.google.common.collect.Sets;
-public class CubeStorageQuery implements ICachableStorageQuery {
+public class CubeStorageQuery implements IStorageQuery {
private static final Logger logger = LoggerFactory.getLogger(CubeStorageQuery.class);
@@ -379,21 +378,4 @@ public class CubeStorageQuery implements ICachableStorageQuery {
}
}
- // ============================================================================
-
- @Override
- public Range<Long> getVolatilePeriod() {
- return null;
- }
-
- @Override
- public String getStorageUUID() {
- return cubeInstance.getUuid();
- }
-
- @Override
- public boolean isDynamic() {
- return false;
- }
-
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/ii/InvertedIndexStorageQuery.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/ii/InvertedIndexStorageQuery.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/ii/InvertedIndexStorageQuery.java
index 5afb62d..fef9662 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/ii/InvertedIndexStorageQuery.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/ii/InvertedIndexStorageQuery.java
@@ -26,19 +26,17 @@ import org.apache.kylin.invertedindex.IISegment;
import org.apache.kylin.metadata.realization.SQLDigest;
import org.apache.kylin.metadata.tuple.ITupleIterator;
import org.apache.kylin.metadata.tuple.TupleInfo;
-import org.apache.kylin.storage.ICachableStorageQuery;
+import org.apache.kylin.storage.IStorageQuery;
import org.apache.kylin.storage.StorageContext;
import org.apache.kylin.storage.hbase.HBaseConnection;
import org.apache.kylin.storage.hbase.ii.coprocessor.endpoint.EndpointTupleIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.collect.Range;
-
/**
* @author yangli9
*/
-public class InvertedIndexStorageQuery implements ICachableStorageQuery {
+public class InvertedIndexStorageQuery implements IStorageQuery {
private static Logger logger = LoggerFactory.getLogger(InvertedIndexStorageQuery.class);
@@ -66,18 +64,4 @@ public class InvertedIndexStorageQuery implements ICachableStorageQuery {
}
}
- @Override
- public Range<Long> getVolatilePeriod() {
- return dataIterator.getCacheExcludedPeriod();
- }
-
- @Override
- public String getStorageUUID() {
- return this.uuid;
- }
-
- @Override
- public boolean isDynamic() {
- return true;
- }
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/1a13952e/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/CubeStorageQueryTest.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/CubeStorageQueryTest.java b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/CubeStorageQueryTest.java
index 8c90abd..71f751b 100644
--- a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/CubeStorageQueryTest.java
+++ b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/CubeStorageQueryTest.java
@@ -41,7 +41,7 @@ import org.apache.kylin.metadata.MetadataManager;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.MeasureDesc;
import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.storage.ICachableStorageQuery;
+import org.apache.kylin.storage.IStorageQuery;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -77,7 +77,7 @@ public class CubeStorageQueryTest extends LocalFileMetadataTestCase {
this.cleanupTestMetadata();
}
- private void validateIdentifyCuboidOnStorageQnery(CubeDesc cubeDesc, ICachableStorageQuery query) {
+ private void validateIdentifyCuboidOnStorageQnery(CubeDesc cubeDesc, IStorageQuery query) {
long baseCuboidId = cubeDesc.getRowkey().getFullMask();
try {
@@ -113,14 +113,14 @@ public class CubeStorageQueryTest extends LocalFileMetadataTestCase {
@Test
public void testIdentifyCuboidV1() {
CubeDesc cubeDesc = cube.getDescriptor();
- ICachableStorageQuery query = new org.apache.kylin.storage.hbase.cube.v1.CubeStorageQuery(cube);
+ IStorageQuery query = new org.apache.kylin.storage.hbase.cube.v1.CubeStorageQuery(cube);
validateIdentifyCuboidOnStorageQnery(cubeDesc, query);
}
@Test
public void testIdentifyCuboidV2() {
CubeDesc cubeDesc = cube.getDescriptor();
- ICachableStorageQuery query = new org.apache.kylin.storage.hbase.cube.v2.CubeStorageQuery(cube);
+ IStorageQuery query = new org.apache.kylin.storage.hbase.cube.v2.CubeStorageQuery(cube);
validateIdentifyCuboidOnStorageQnery(cubeDesc, query);
}
[3/4] kylin git commit: KYLIN-1453 cuboid sharding based on UHC column
Posted by ma...@apache.org.
KYLIN-1453 cuboid sharding based on UHC column
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/dfab0c14
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/dfab0c14
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/dfab0c14
Branch: refs/heads/master
Commit: dfab0c14cf77c0c77687d66858e68500ae3776ab
Parents: 1a13952
Author: Hongbin Ma <ma...@apache.org>
Authored: Fri Mar 18 11:26:01 2016 +0800
Committer: Hongbin Ma <ma...@apache.org>
Committed: Thu Mar 24 14:45:02 2016 +0800
----------------------------------------------------------------------
.../java/org/apache/kylin/cube/CubeSegment.java | 33 +--
.../org/apache/kylin/cube/kv/RowKeyEncoder.java | 19 +-
.../org/apache/kylin/cube/model/CubeDesc.java | 5 +
.../apache/kylin/cube/model/RowKeyColDesc.java | 243 ++++++++++---------
.../org/apache/kylin/cube/model/RowKeyDesc.java | 13 +
5 files changed, 181 insertions(+), 132 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/dfab0c14/core-cube/src/main/java/org/apache/kylin/cube/CubeSegment.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/CubeSegment.java b/core-cube/src/main/java/org/apache/kylin/cube/CubeSegment.java
index ead6a41..8dd294a 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/CubeSegment.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/CubeSegment.java
@@ -18,12 +18,14 @@
package org.apache.kylin.cube;
-import com.fasterxml.jackson.annotation.JsonAutoDetect;
-import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
-import com.fasterxml.jackson.annotation.JsonBackReference;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
+import java.text.SimpleDateFormat;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TimeZone;
+import java.util.concurrent.ConcurrentHashMap;
+
import org.apache.kylin.common.persistence.ResourceStore;
import org.apache.kylin.common.util.ShardingHash;
import org.apache.kylin.cube.kv.CubeDimEncMap;
@@ -37,12 +39,12 @@ import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.metadata.realization.IRealization;
import org.apache.kylin.metadata.realization.IRealizationSegment;
-import java.text.SimpleDateFormat;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.TimeZone;
-import java.util.concurrent.ConcurrentHashMap;
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
+import com.fasterxml.jackson.annotation.JsonBackReference;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
@JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
public class CubeSegment implements Comparable<CubeSegment>, IRealizationSegment {
@@ -80,7 +82,6 @@ public class CubeSegment implements Comparable<CubeSegment>, IRealizationSegment
@JsonProperty("blackout_cuboids")
private List<Long> blackoutCuboids = Lists.newArrayList();
-
@JsonProperty("binary_signature")
private String binarySignature; // a hash of cube schema and dictionary ID, used for sanity check
@@ -280,7 +281,7 @@ public class CubeSegment implements Comparable<CubeSegment>, IRealizationSegment
public Dictionary<String> getDictionary(TblColRef col) {
return CubeManager.getInstance(this.getCubeInstance().getConfig()).getDictionary(this, col);
}
-
+
public CubeDimEncMap getDimensionEncodingMap() {
return new CubeDimEncMap(this);
}
@@ -382,6 +383,10 @@ public class CubeSegment implements Comparable<CubeSegment>, IRealizationSegment
return getCubeDesc().isEnableSharding();
}
+ public Set<TblColRef> getUHCColumns() {
+ return getCubeDesc().getUHCColumns();
+ }
+
public int getRowKeyPreambleSize() {
return isEnableSharding() ? RowConstants.ROWKEY_SHARD_AND_CUBOID_LEN : RowConstants.ROWKEY_CUBOIDID_LEN;
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/dfab0c14/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoder.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoder.java b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoder.java
index 05afdbf..656ddc1 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoder.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowKeyEncoder.java
@@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.Bytes;
@@ -39,13 +40,24 @@ public class RowKeyEncoder extends AbstractRowKeyEncoder {
private int bodyLength = 0;
private RowKeyColumnIO colIO;
+
protected boolean enableSharding;
+ private int UHCOffset = -1;//it's a offset to the beginning of body
+ private int UHCLength = -1;
public RowKeyEncoder(CubeSegment cubeSeg, Cuboid cuboid) {
super(cubeSeg, cuboid);
enableSharding = cubeSeg.isEnableSharding();
+ Set<TblColRef> UHCColumns = cubeSeg.getUHCColumns();
+ if (UHCColumns.size() > 1) {
+ throw new IllegalStateException("Does not support multiple UHC now");
+ }
colIO = new RowKeyColumnIO(cubeSeg.getDimensionEncodingMap());
for (TblColRef column : cuboid.getColumns()) {
+ if (UHCColumns.contains(column)) {
+ UHCOffset = bodyLength;
+ UHCLength = colIO.getColumnLength(column);
+ }
bodyLength += colIO.getColumnLength(column);
}
}
@@ -60,9 +72,10 @@ public class RowKeyEncoder extends AbstractRowKeyEncoder {
protected short calculateShard(byte[] key) {
if (enableSharding) {
- int bodyOffset = RowConstants.ROWKEY_SHARD_AND_CUBOID_LEN;
+ int shardSeedOffset = UHCOffset == -1 ? 0 : UHCOffset;
+ int shardSeedLength = UHCLength == -1 ? bodyLength : UHCLength;
short cuboidShardNum = cubeSeg.getCuboidShardNum(cuboid.getId());
- short shardOffset = ShardingHash.getShard(key, bodyOffset, bodyLength, cuboidShardNum);
+ short shardOffset = ShardingHash.getShard(key, RowConstants.ROWKEY_SHARD_AND_CUBOID_LEN + shardSeedOffset, shardSeedLength, cuboidShardNum);
return ShardingHash.normalize(cubeSeg.getCuboidBaseShard(cuboid.getId()), shardOffset, cubeSeg.getTotalShards());
} else {
throw new RuntimeException("If enableSharding false, you should never calculate shard");
@@ -90,7 +103,7 @@ public class RowKeyEncoder extends AbstractRowKeyEncoder {
@Override
public void encode(ByteArray bodyBytes, ByteArray outputBuf) {
Preconditions.checkState(bodyBytes.length() == bodyLength);
- Preconditions.checkState(bodyBytes.length() + getHeaderLength() == outputBuf.length(),//
+ Preconditions.checkState(bodyBytes.length() + getHeaderLength() == outputBuf.length(), //
"bodybytes length: " + bodyBytes.length() + " outputBuf length: " + outputBuf.length() + " header length: " + getHeaderLength());
System.arraycopy(bodyBytes.array(), bodyBytes.offset(), outputBuf.array(), getHeaderLength(), bodyLength);
http://git-wip-us.apache.org/repos/asf/kylin/blob/dfab0c14/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
index cb94447..165ab60 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/CubeDesc.java
@@ -166,6 +166,11 @@ public class CubeDesc extends RootPersistentEntity {
return storageType == IStorageAware.ID_SHARDED_HBASE;
}
+ public Set<TblColRef> getUHCColumns() {
+ return getRowkey().getUHCColumns();
+ }
+
+
/**
* Error messages during resolving json metadata
*/
http://git-wip-us.apache.org/repos/asf/kylin/blob/dfab0c14/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 452d689..e72011c 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
@@ -1,115 +1,128 @@
-/*
- * 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.model;
-
-import org.apache.commons.lang.StringUtils;
-import org.apache.kylin.common.util.StringUtil;
-import org.apache.kylin.dimension.DictionaryDimEnc;
-import org.apache.kylin.dimension.DimensionEncodingFactory;
-import org.apache.kylin.metadata.model.TblColRef;
-
-import com.fasterxml.jackson.annotation.JsonAutoDetect;
-import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.google.common.base.Objects;
-import com.google.common.base.Preconditions;
-
-/**
- * @author yangli9
- *
- */
-@JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
-public class RowKeyColDesc {
-
- @JsonProperty("column")
- private String column;
- @JsonProperty("encoding")
- private String encoding;
-
- // computed
- private String encodingName;
- private String[] encodingArgs;
- private int bitIndex;
- private TblColRef colRef;
-
- public void init() {
- Preconditions.checkState(StringUtils.isNotEmpty(this.encoding));
-
- String[] parts = this.encoding.split("\\s*[(),:]\\s*");
- if (parts == null || parts.length == 0 || parts[0].isEmpty())
- throw new IllegalArgumentException("Not supported row key col encoding: '" + this.encoding + "'");
-
- this.encodingName = parts[0];
- this.encodingArgs = parts[parts.length - 1].isEmpty() //
- ? StringUtil.subArray(parts, 1, parts.length - 1) : StringUtil.subArray(parts, 1, parts.length);
-
- if (!DimensionEncodingFactory.isVaildEncoding(this.encodingName))
- throw new IllegalArgumentException("Not supported row key col encoding: '" + this.encoding + "'");
- }
-
- public String getEncoding() {
- return encoding;
- }
-
- public void setEncoding(String encoding) {
- this.encoding = encoding;
- }
-
- public String getColumn() {
- return column;
- }
-
- public void setColumn(String column) {
- this.column = column;
- }
-
- public String getEncodingName() {
- return encodingName;
- }
-
- public String[] getEncodingArgs() {
- return encodingArgs;
- }
-
- public boolean isUsingDictionary() {
- return DictionaryDimEnc.ENCODING_NAME.equals(encodingName);
- }
-
- public int getBitIndex() {
- return bitIndex;
- }
-
- void setBitIndex(int index) {
- this.bitIndex = index;
- }
-
- public TblColRef getColRef() {
- return colRef;
- }
-
- void setColRef(TblColRef colRef) {
- this.colRef = colRef;
- }
-
- @Override
- public String toString() {
- return Objects.toStringHelper(this).add("column", column).add("encoding", encoding).toString();
- }
-
-}
+/*
+ * 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.model;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.kylin.common.util.StringUtil;
+import org.apache.kylin.dimension.DictionaryDimEnc;
+import org.apache.kylin.dimension.DimensionEncodingFactory;
+import org.apache.kylin.metadata.model.TblColRef;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Objects;
+import com.google.common.base.Preconditions;
+
+/**
+ * @author yangli9
+ *
+ */
+@JsonAutoDetect(fieldVisibility = Visibility.NONE, getterVisibility = Visibility.NONE, isGetterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE)
+public class RowKeyColDesc {
+
+ @JsonProperty("column")
+ private String column;
+ @JsonProperty("encoding")
+ private String encoding;
+ @JsonProperty("isUHC")
+ private boolean isUHC;//is ultra high cardinality column
+
+
+ // computed
+ private String encodingName;
+ private String[] encodingArgs;
+ private int bitIndex;
+ private TblColRef colRef;
+
+ public void init() {
+ Preconditions.checkState(StringUtils.isNotEmpty(this.encoding));
+
+ String[] parts = this.encoding.split("\\s*[(),:]\\s*");
+ if (parts == null || parts.length == 0 || parts[0].isEmpty())
+ throw new IllegalArgumentException("Not supported row key col encoding: '" + this.encoding + "'");
+
+ this.encodingName = parts[0];
+ this.encodingArgs = parts[parts.length - 1].isEmpty() //
+ ? StringUtil.subArray(parts, 1, parts.length - 1) : StringUtil.subArray(parts, 1, parts.length);
+
+ if (!DimensionEncodingFactory.isVaildEncoding(this.encodingName))
+ throw new IllegalArgumentException("Not supported row key col encoding: '" + this.encoding + "'");
+ }
+
+ public String getEncoding() {
+ return encoding;
+ }
+
+ public void setEncoding(String encoding) {
+ this.encoding = encoding;
+ }
+
+ public String getColumn() {
+ return column;
+ }
+
+ public void setColumn(String column) {
+ this.column = column;
+ }
+
+ public boolean isUHC() {
+ return isUHC;
+ }
+
+ public void setUHC(boolean UHC) {
+ isUHC = UHC;
+ }
+
+
+
+ public String getEncodingName() {
+ return encodingName;
+ }
+
+ public String[] getEncodingArgs() {
+ return encodingArgs;
+ }
+
+ public boolean isUsingDictionary() {
+ return DictionaryDimEnc.ENCODING_NAME.equals(encodingName);
+ }
+
+ public int getBitIndex() {
+ return bitIndex;
+ }
+
+ void setBitIndex(int index) {
+ this.bitIndex = index;
+ }
+
+ public TblColRef getColRef() {
+ return colRef;
+ }
+
+ void setColRef(TblColRef colRef) {
+ this.colRef = colRef;
+ }
+
+ @Override
+ public String toString() {
+ return Objects.toStringHelper(this).add("column", column).add("encoding", encoding).toString();
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kylin/blob/dfab0c14/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyDesc.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyDesc.java b/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyDesc.java
index 6052c08..2df1764 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyDesc.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/model/RowKeyDesc.java
@@ -20,7 +20,9 @@ package org.apache.kylin.cube.model;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
+import java.util.Set;
import org.apache.kylin.metadata.model.TblColRef;
@@ -41,6 +43,7 @@ public class RowKeyDesc {
private long fullMask;
private CubeDesc cubeDesc;
private Map<TblColRef, RowKeyColDesc> columnMap;
+ private Set<TblColRef> UHCColumns;
public RowKeyColDesc[] getRowKeyColumns() {
return rowkeyColumns;
@@ -65,6 +68,10 @@ public class RowKeyDesc {
return getColDesc(col).isUsingDictionary();
}
+ public Set<TblColRef> getUHCColumns() {
+ return UHCColumns;
+ }
+
public void init(CubeDesc cubeDesc) {
setCubeDesc(cubeDesc);
@@ -84,6 +91,7 @@ public class RowKeyDesc {
private void buildRowKey(Map<String, TblColRef> colNameAbbr) {
columnMap = new HashMap<TblColRef, RowKeyColDesc>();
+ UHCColumns = new HashSet<>();
for (int i = 0; i < rowkeyColumns.length; i++) {
RowKeyColDesc rowKeyColDesc = rowkeyColumns[i];
@@ -97,6 +105,11 @@ public class RowKeyDesc {
}
columnMap.put(rowKeyColDesc.getColRef(), rowKeyColDesc);
+
+ if (rowKeyColDesc.isUHC()) {
+ UHCColumns.add(rowKeyColDesc.getColRef());
+
+ }
}
this.fullMask = 0L;