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/01/21 16:33:21 UTC
[12/13] kylin git commit: KYLIN-1356 move all it to a seprate module
http://git-wip-us.apache.org/repos/asf/kylin/blob/7f08dab5/assembly/src/test/java/org/apache/kylin/source/hive/ITHiveTableReaderTest.java
----------------------------------------------------------------------
diff --git a/assembly/src/test/java/org/apache/kylin/source/hive/ITHiveTableReaderTest.java b/assembly/src/test/java/org/apache/kylin/source/hive/ITHiveTableReaderTest.java
deleted file mode 100644
index 57c0be3..0000000
--- a/assembly/src/test/java/org/apache/kylin/source/hive/ITHiveTableReaderTest.java
+++ /dev/null
@@ -1,49 +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.source.hive;
-
-import java.io.IOException;
-
-import org.apache.kylin.storage.hbase.steps.HBaseMetadataTestCase;
-import org.junit.Assert;
-import org.junit.Test;
-
-/**
- * This test case need the hive runtime; Please run it with sandbox;
- * @author shaoshi
- *
- * It is in the exclude list of default profile in pom.xml
- */
-public class ITHiveTableReaderTest extends HBaseMetadataTestCase {
-
- @Test
- public void test() throws IOException {
- HiveTableReader reader = new HiveTableReader("default", "test_kylin_fact");
- int rowNumber = 0;
- while (reader.next()) {
- String[] row = reader.getRow();
- Assert.assertEquals(9, row.length);
- //System.out.println(ArrayUtils.toString(row));
- rowNumber++;
- }
-
- reader.close();
- Assert.assertEquals(10000, rowNumber);
- }
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/7f08dab5/assembly/src/test/java/org/apache/kylin/source/hive/ITSnapshotManagerTest.java
----------------------------------------------------------------------
diff --git a/assembly/src/test/java/org/apache/kylin/source/hive/ITSnapshotManagerTest.java b/assembly/src/test/java/org/apache/kylin/source/hive/ITSnapshotManagerTest.java
deleted file mode 100644
index 0df632a..0000000
--- a/assembly/src/test/java/org/apache/kylin/source/hive/ITSnapshotManagerTest.java
+++ /dev/null
@@ -1,83 +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.source.hive;
-
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertEquals;
-
-import org.apache.kylin.dict.lookup.SnapshotManager;
-import org.apache.kylin.dict.lookup.SnapshotTable;
-import org.apache.kylin.metadata.MetadataManager;
-import org.apache.kylin.metadata.model.TableDesc;
-import org.apache.kylin.source.ReadableTable;
-import org.apache.kylin.source.ReadableTable.TableReader;
-import org.apache.kylin.source.SourceFactory;
-import org.apache.kylin.storage.hbase.steps.HBaseMetadataTestCase;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-/**
- * @author yangli9
- *
- */
-public class ITSnapshotManagerTest extends HBaseMetadataTestCase {
-
- SnapshotManager snapshotMgr;
-
- @Before
- public void setup() throws Exception {
- createTestMetadata();
- snapshotMgr = SnapshotManager.getInstance(getTestConfig());
- }
-
- @After
- public void after() throws Exception {
- cleanupTestMetadata();
- }
-
- @Test
- public void basicTest() throws Exception {
- String tableName = "EDW.TEST_SITES";
- TableDesc tableDesc = MetadataManager.getInstance(getTestConfig()).getTableDesc(tableName);
- ReadableTable hiveTable = SourceFactory.createReadableTable(tableDesc);
- String snapshotPath = snapshotMgr.buildSnapshot(hiveTable, tableDesc).getResourcePath();
-
- snapshotMgr.wipeoutCache();
-
- SnapshotTable snapshot = snapshotMgr.getSnapshotTable(snapshotPath);
-
- // compare hive & snapshot
- TableReader hiveReader = hiveTable.getReader();
- TableReader snapshotReader = snapshot.getReader();
-
- while (true) {
- boolean hiveNext = hiveReader.next();
- boolean snapshotNext = snapshotReader.next();
- assertEquals(hiveNext, snapshotNext);
-
- if (hiveNext == false)
- break;
-
- String[] hiveRow = hiveReader.getRow();
- String[] snapshotRow = snapshotReader.getRow();
- assertArrayEquals(hiveRow, snapshotRow);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/7f08dab5/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 ae96e67..b481557 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
@@ -79,6 +79,8 @@ public class BasicTest {
@Test
public void testxx() throws InterruptedException {
+ System.out.println(System.getProperty("skipTests").length());
+ System.out.println(System.getProperty("skipXests"));
byte[][] data = new byte[10000000][];
byte[] temp = new byte[100];
for (int i = 0; i < 100; i++) {
http://git-wip-us.apache.org/repos/asf/kylin/blob/7f08dab5/core-common/src/test/java/org/apache/kylin/common/util/HBaseMetadataTestCase.java
----------------------------------------------------------------------
diff --git a/core-common/src/test/java/org/apache/kylin/common/util/HBaseMetadataTestCase.java b/core-common/src/test/java/org/apache/kylin/common/util/HBaseMetadataTestCase.java
new file mode 100644
index 0000000..9d78957
--- /dev/null
+++ b/core-common/src/test/java/org/apache/kylin/common/util/HBaseMetadataTestCase.java
@@ -0,0 +1,63 @@
+/*
+ * 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.common.util;
+
+import java.io.File;
+
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.AbstractKylinTestCase;
+import org.apache.kylin.common.util.ClassUtil;
+
+/**
+ * @author ysong1
+ */
+public class HBaseMetadataTestCase extends AbstractKylinTestCase {
+
+ static {
+ try {
+ ClassUtil.addClasspath(new File("../examples/test_case_data/sandbox/").getAbsolutePath());
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ @Override
+ public void createTestMetadata() throws Exception {
+ staticCreateTestMetadata();
+ }
+
+ @Override
+ public void cleanupTestMetadata() {
+ staticCleanupTestMetadata();
+ }
+
+ public static void staticCreateTestMetadata() throws Exception {
+ staticCreateTestMetadata(SANDBOX_TEST_DATA);
+ }
+
+ public static void staticCreateTestMetadata(String kylinConfigFolder) {
+
+ KylinConfig.destoryInstance();
+
+ if (System.getProperty(KylinConfig.KYLIN_CONF) == null && System.getenv(KylinConfig.KYLIN_CONF) == null)
+ System.setProperty(KylinConfig.KYLIN_CONF, kylinConfigFolder);
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/7f08dab5/kylin-it/pom.xml
----------------------------------------------------------------------
diff --git a/kylin-it/pom.xml b/kylin-it/pom.xml
new file mode 100644
index 0000000..cc3e9b2
--- /dev/null
+++ b/kylin-it/pom.xml
@@ -0,0 +1,367 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>kylin</artifactId>
+ <groupId>org.apache.kylin</groupId>
+ <version>2.0-incubating-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>kylin-it</artifactId>
+ <name>Kylin:IT</name>
+
+
+ <!-- Dependencies. -->
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.kylin</groupId>
+ <artifactId>atopcalcite</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kylin</groupId>
+ <artifactId>kylin-core-storage</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kylin</groupId>
+ <artifactId>kylin-source-hive</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kylin</groupId>
+ <artifactId>kylin-jdbc</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kylin</groupId>
+ <artifactId>kylin-storage-hbase</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kylin</groupId>
+ <artifactId>kylin-query</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kylin</groupId>
+ <artifactId>kylin-invertedindex</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.calcite</groupId>
+ <artifactId>calcite-linq4j</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>commons-cli</groupId>
+ <artifactId>commons-cli</artifactId>
+ </dependency>
+
+ <!-- Env & Test -->
+
+ <dependency>
+ <groupId>org.apache.kylin</groupId>
+ <artifactId>kylin-core-common</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kylin</groupId>
+ <artifactId>kylin-core-storage</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kylin</groupId>
+ <artifactId>kylin-storage-hbase</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kylin</groupId>
+ <artifactId>kylin-storage-hbase</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kylin</groupId>
+ <artifactId>kylin-assembly</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>xerces</groupId>
+ <artifactId>xercesImpl</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>xalan</groupId>
+ <artifactId>xalan</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.h2database</groupId>
+ <artifactId>h2</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.dbunit</groupId>
+ <artifactId>dbunit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-mapreduce-client-core</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs</artifactId>
+ <scope>provided</scope>
+ <!-- protobuf version conflict with hbase -->
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-java</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-client</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-server</artifactId>
+ <scope>provided</scope>
+ <!-- version conflict with hadoop2.2 -->
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-jdbc</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-testing-util</artifactId>
+ <version>${hbase-hadoop2.version}</version>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>javax.servlet</groupId>
+ <artifactId>servlet-api</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>javax.servlet.jsp</groupId>
+ <artifactId>jsp-api</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ </dependencies>
+
+
+ <profiles>
+ <profile>
+ <id>sandbox</id>
+ <activation>
+ <activeByDefault>true</activeByDefault>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <version>2.16</version>
+ <executions>
+ <execution>
+ <id>integration-tests</id>
+ <phase>integration-test</phase>
+ <goals>
+ <goal>integration-test</goal>
+ <goal>verify</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <excludes>
+ <exclude>**/*$*</exclude>
+ </excludes>
+ <systemProperties>
+ <property>
+ <name>useSandbox</name>
+ <value>true</value>
+ </property>
+ </systemProperties>
+ </configuration>
+ </plugin>
+
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>exec-maven-plugin</artifactId>
+ <version>1.4.0</version>
+ <executions>
+ <execution>
+ <id>build_cube_with_engine</id>
+ <goals>
+ <goal>java</goal>
+ </goals>
+ <phase>pre-integration-test</phase>
+ <configuration>
+ <skip>${skipTests}</skip>
+ <classpathScope>test</classpathScope>
+ <mainClass>org.apache.kylin.provision.BuildCubeWithEngine</mainClass>
+ </configuration>
+ </execution>
+ <execution>
+ <id>build_cube_with_stream</id>
+ <goals>
+ <goal>java</goal>
+ </goals>
+ <phase>pre-integration-test</phase>
+ <configuration>
+ <skip>${skipTests}</skip>
+ <classpathScope>test</classpathScope>
+ <mainClass>org.apache.kylin.provision.BuildCubeWithStream</mainClass>
+ </configuration>
+ </execution>
+ <execution>
+ <id>build_ii_with_stream</id>
+ <goals>
+ <goal>java</goal>
+ </goals>
+ <phase>pre-integration-test</phase>
+ <configuration>
+ <skip>${skipTests}</skip>
+ <classpathScope>test</classpathScope>
+ <mainClass>org.apache.kylin.provision.BuildIIWithStream</mainClass>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ </plugins>
+ </build>
+ </profile>
+ <profile>
+ <!-- This profile adds/overrides few features of the 'apache-release'
+ profile in the parent pom. -->
+ <id>apache-release</id>
+ <build>
+ <plugins>
+ <!-- Apache-RAT checks for files without headers.
+ If run on a messy developer's sandbox, it will fail.
+ This serves as a reminder to only build a release in a clean
+ sandbox! -->
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <configuration>
+ <numUnapprovedLicenses>0</numUnapprovedLicenses>
+ <excludes>
+ <!-- test data -->
+ <exclude>src/test/**/*.dat</exclude>
+ <exclude>src/test/**/*.expected</exclude>
+ </excludes>
+ </configuration>
+ <executions>
+ <execution>
+ <phase>verify</phase>
+ <goals>
+ <goal>check</goal>
+ </goals>
+ </execution>
+ </executions>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.maven.doxia</groupId>
+ <artifactId>doxia-core</artifactId>
+ <version>1.6</version>
+ <exclusions>
+ <exclusion>
+ <groupId>xerces</groupId>
+ <artifactId>xercesImpl</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ </dependencies>
+ </plugin>
+ <plugin>
+ <groupId>net.ju-n.maven.plugins</groupId>
+ <artifactId>checksum-maven-plugin</artifactId>
+ <version>1.2</version>
+ <executions>
+ <execution>
+ <goals>
+ <goal>artifacts</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <algorithms>
+ <algorithm>MD5</algorithm>
+ <algorithm>SHA-1</algorithm>
+ </algorithms>
+ <failOnError>false</failOnError>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+
+</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kylin/blob/7f08dab5/kylin-it/src/test/java/org/apache/kylin/jdbc/ITJDBCDriverTest.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/jdbc/ITJDBCDriverTest.java b/kylin-it/src/test/java/org/apache/kylin/jdbc/ITJDBCDriverTest.java
new file mode 100644
index 0000000..9b67248
--- /dev/null
+++ b/kylin-it/src/test/java/org/apache/kylin/jdbc/ITJDBCDriverTest.java
@@ -0,0 +1,281 @@
+/*
+ * 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.jdbc;
+
+import java.io.File;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Properties;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.kylin.common.util.HBaseMetadataTestCase;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.webapp.WebAppContext;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
+
+/**
+ */
+public class ITJDBCDriverTest extends HBaseMetadataTestCase {
+
+ private static Server server = null;
+ private static SystemPropertiesOverride sysPropsOverride = new SystemPropertiesOverride();
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ sysPropsOverride.override("spring.profiles.active", "testing");
+ sysPropsOverride.override("catalina.home", "."); // resources/log4j.properties ref ${catalina.home}
+ staticCreateTestMetadata();
+ startJetty();
+ }
+
+ @AfterClass
+ public static void afterClass() throws Exception {
+ stopJetty();
+ staticCleanupTestMetadata();
+ sysPropsOverride.restore();
+ }
+
+ protected static void stopJetty() throws Exception {
+ if (server != null)
+ server.stop();
+
+ File workFolder = new File("work");
+ if (workFolder.isDirectory() && workFolder.exists()) {
+ FileUtils.deleteDirectory(workFolder);
+ }
+ }
+
+ protected static void startJetty() throws Exception {
+
+ server = new Server(7070);
+
+ WebAppContext context = new WebAppContext();
+ context.setDescriptor("./src/main/webapp/WEB-INF/web.xml");
+ context.setResourceBase("./src/main/webapp");
+ context.setContextPath("/kylin");
+ context.setParentLoaderPriority(true);
+
+ server.setHandler(context);
+
+ server.start();
+
+ }
+
+ protected Connection getConnection() throws Exception {
+
+ Driver driver = (Driver) Class.forName("org.apache.kylin.jdbc.Driver").newInstance();
+ Properties info = new Properties();
+ info.put("user", "ADMIN");
+ info.put("password", "KYLIN");
+ Connection conn = driver.connect("jdbc:kylin://localhost:7070/default", info);
+
+ return conn;
+ }
+
+ @Test
+ public void testMetadata1() throws Exception {
+
+ // check the JDBC API here: http://docs.oracle.com/javase/7/docs/api/java/sql/DatabaseMetaData.html
+ Connection conn = getConnection();
+
+ // test getSchemas();
+ List<String> schemaList = Lists.newArrayList();
+ DatabaseMetaData dbMetadata = conn.getMetaData();
+ ResultSet resultSet = dbMetadata.getSchemas();
+ while (resultSet.next()) {
+ String schema = resultSet.getString("TABLE_SCHEM");
+ String catalog = resultSet.getString("TABLE_CATALOG");
+
+ System.out.println("Get schema: schema=" + schema + ", catalog=" + catalog);
+ schemaList.add(schema);
+
+ }
+
+ resultSet.close();
+ Assert.assertTrue(schemaList.contains("DEFAULT"));
+ Assert.assertTrue(schemaList.contains("EDW"));
+
+ // test getCatalogs();
+ resultSet = dbMetadata.getCatalogs();
+
+ List<String> catalogList = Lists.newArrayList();
+ while (resultSet.next()) {
+ String catalog = resultSet.getString("TABLE_CAT");
+
+ System.out.println("Get catalog: catalog=" + catalog);
+ catalogList.add(catalog);
+
+ }
+ Assert.assertTrue(catalogList.size() > 0 && catalogList.contains("defaultCatalog"));
+
+ /** //Disable the test on getTableTypes() as it is not ready
+ resultSet = dbMetadata.getTableTypes();
+
+ List<String> tableTypes = Lists.newArrayList();
+ while (resultSet.next()) {
+ String type = resultSet.getString("TABLE_TYPE");
+
+ System.out.println("Get table type: type=" + type);
+ tableTypes.add(type);
+
+ }
+
+ Assert.assertTrue(tableTypes.size() > 0 && tableTypes.contains("TABLE"));
+ resultSet.close();
+
+ **/
+ conn.close();
+ }
+
+ @Test
+ public void testMetadata2() throws Exception {
+ Connection conn = getConnection();
+
+ List<String> tableList = Lists.newArrayList();
+ DatabaseMetaData dbMetadata = conn.getMetaData();
+ ResultSet resultSet = dbMetadata.getTables(null, "%", "%", new String[] { "TABLE" });
+ while (resultSet.next()) {
+ String schema = resultSet.getString("TABLE_SCHEM");
+ String name = resultSet.getString("TABLE_NAME");
+
+ System.out.println("Get table: schema=" + schema + ", name=" + name);
+ tableList.add(schema + "." + name);
+
+ }
+
+ resultSet.close();
+ Assert.assertTrue(tableList.contains("DEFAULT.TEST_KYLIN_FACT"));
+
+ resultSet = dbMetadata.getColumns(null, "%", "TEST_KYLIN_FACT", "%");
+
+ List<String> columns = Lists.newArrayList();
+ while (resultSet.next()) {
+ String name = resultSet.getString("COLUMN_NAME");
+ String type = resultSet.getString("TYPE_NAME");
+
+ System.out.println("Get column: name=" + name + ", data_type=" + type);
+ columns.add(name);
+
+ }
+
+ Assert.assertTrue(columns.size() > 0 && columns.contains("CAL_DT"));
+ resultSet.close();
+ conn.close();
+ }
+
+ @Test
+ public void testSimpleStatement() throws Exception {
+ Connection conn = getConnection();
+ Statement statement = conn.createStatement();
+
+ statement.execute("select count(*) from test_kylin_fact");
+
+ ResultSet rs = statement.getResultSet();
+
+ Assert.assertTrue(rs.next());
+ int result = rs.getInt(1);
+
+ Assert.assertTrue(result > 0);
+
+ rs.close();
+ statement.close();
+ conn.close();
+
+ }
+
+ @Test
+ public void testPreparedStatement() throws Exception {
+ Connection conn = getConnection();
+
+ PreparedStatement statement = conn.prepareStatement("select LSTG_FORMAT_NAME, sum(price) as GMV, count(1) as TRANS_CNT from test_kylin_fact " + "where LSTG_FORMAT_NAME = ? group by LSTG_FORMAT_NAME");
+
+ statement.setString(1, "FP-GTC");
+
+ ResultSet rs = statement.executeQuery();
+
+ Assert.assertTrue(rs.next());
+
+ String format_name = rs.getString(1);
+
+ Assert.assertTrue("FP-GTC".equals(format_name));
+
+ rs.close();
+ statement.close();
+ conn.close();
+
+ }
+
+ @Test
+ public void testResultSet() throws Exception {
+ String sql = "select LSTG_FORMAT_NAME, sum(price) as GMV, count(1) as TRANS_CNT from test_kylin_fact \n" + " group by LSTG_FORMAT_NAME ";
+
+ Connection conn = getConnection();
+ Statement statement = conn.createStatement();
+
+ statement.execute(sql);
+
+ ResultSet rs = statement.getResultSet();
+
+ int count = 0;
+ while (rs.next()) {
+ count++;
+ String lstg = rs.getString(1);
+ double gmv = rs.getDouble(2);
+ int trans_count = rs.getInt(3);
+
+ System.out.println("Get a line: LSTG_FORMAT_NAME=" + lstg + ", GMV=" + gmv + ", TRANS_CNT=" + trans_count);
+ }
+
+ Assert.assertTrue(count > 0);
+ statement.close();
+ rs.close();
+ conn.close();
+
+ }
+
+ private static class SystemPropertiesOverride {
+ HashMap<String, String> backup = new HashMap<String, String>();
+
+ public void override(String key, String value) {
+ backup.put(key, System.getProperty(key));
+ System.setProperty(key, value);
+ }
+
+ public void restore() {
+ for (String key : backup.keySet()) {
+ String value = backup.get(key);
+ if (value == null)
+ System.clearProperty(key);
+ else
+ System.setProperty(key, value);
+ }
+ backup.clear();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/7f08dab5/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithEngine.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithEngine.java b/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithEngine.java
new file mode 100644
index 0000000..c40b470
--- /dev/null
+++ b/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithEngine.java
@@ -0,0 +1,334 @@
+/*
+ * 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.provision;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.File;
+import java.lang.reflect.Method;
+import java.text.SimpleDateFormat;
+import java.util.List;
+import java.util.TimeZone;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.AbstractKylinTestCase;
+import org.apache.kylin.common.util.ClassUtil;
+import org.apache.kylin.common.util.HBaseMetadataTestCase;
+import org.apache.kylin.cube.CubeInstance;
+import org.apache.kylin.cube.CubeManager;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.cube.CubeUpdate;
+import org.apache.kylin.engine.EngineFactory;
+import org.apache.kylin.engine.mr.CubingJob;
+import org.apache.kylin.job.DeployUtil;
+import org.apache.kylin.job.engine.JobEngineConfig;
+import org.apache.kylin.job.execution.AbstractExecutable;
+import org.apache.kylin.job.execution.DefaultChainedExecutable;
+import org.apache.kylin.job.execution.ExecutableState;
+import org.apache.kylin.job.impl.threadpool.DefaultScheduler;
+import org.apache.kylin.job.manager.ExecutableManager;
+import org.apache.kylin.storage.hbase.util.StorageCleanupJob;
+import org.apache.kylin.storage.hbase.util.ZookeeperJobLock;
+
+import com.google.common.collect.Lists;
+
+public class BuildCubeWithEngine {
+
+ private CubeManager cubeManager;
+ private DefaultScheduler scheduler;
+ protected ExecutableManager jobService;
+ private static boolean fastBuildMode = false;
+
+ private static final Log logger = LogFactory.getLog(BuildCubeWithEngine.class);
+
+ public static void main(String[] args) throws Exception {
+ beforeClass();
+ BuildCubeWithEngine buildCubeWithEngine = new BuildCubeWithEngine();
+ buildCubeWithEngine.before();
+ buildCubeWithEngine.build();
+ afterClass();
+ }
+
+ public static void beforeClass() throws Exception {
+ logger.info("Adding to classpath: " + new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+ ClassUtil.addClasspath(new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+
+ String fastModeStr = System.getProperty("fastBuildMode");
+ if (fastModeStr != null && fastModeStr.equalsIgnoreCase("true")) {
+ fastBuildMode = true;
+ logger.info("Will use fast build mode");
+ } else {
+ logger.info("Will not use fast build mode");
+ }
+
+ System.setProperty(KylinConfig.KYLIN_CONF, "../examples/test_case_data/sandbox");
+ if (System.getProperty("hdp.version") == null) {
+ throw new RuntimeException("No hdp.version set; Please set hdp.version in your jvm option, for example: -Dhdp.version=2.2.4.2-2");
+ }
+
+ HBaseMetadataTestCase.staticCreateTestMetadata(AbstractKylinTestCase.SANDBOX_TEST_DATA);
+ }
+
+ public void before() throws Exception {
+
+ DeployUtil.initCliWorkDir();
+ DeployUtil.deployMetadata();
+ DeployUtil.overrideJobJarLocations();
+
+ final KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
+ jobService = ExecutableManager.getInstance(kylinConfig);
+ scheduler = DefaultScheduler.getInstance();
+ scheduler.init(new JobEngineConfig(kylinConfig), new ZookeeperJobLock());
+ if (!scheduler.hasStarted()) {
+ throw new RuntimeException("scheduler has not been started");
+ }
+ cubeManager = CubeManager.getInstance(kylinConfig);
+ for (String jobId : jobService.getAllJobIds()) {
+ if (jobService.getJob(jobId) instanceof CubingJob) {
+ jobService.deleteJob(jobId);
+ }
+ }
+
+ }
+
+ public static void afterClass() {
+ HBaseMetadataTestCase.staticCleanupTestMetadata();
+ }
+
+ public void build() throws Exception {
+ DeployUtil.prepareTestDataForNormalCubes("test_kylin_cube_with_slr_left_join_empty");
+ testInner();
+ testLeft();
+ }
+
+ protected void waitForJob(String jobId) {
+ while (true) {
+ AbstractExecutable job = jobService.getJob(jobId);
+ if (job.getStatus() == ExecutableState.SUCCEED || job.getStatus() == ExecutableState.ERROR) {
+ break;
+ } else {
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+ }
+
+ private void testInner() throws Exception {
+ String[] testCase = new String[] { "testInnerJoinCubeWithoutSlr", "testInnerJoinCubeWithSlr" };
+ runTestAndAssertSucceed(testCase);
+ }
+
+ private void testLeft() throws Exception {
+ String[] testCase = new String[] { "testLeftJoinCubeWithSlr", "testLeftJoinCubeWithoutSlr" };
+ runTestAndAssertSucceed(testCase);
+ }
+
+ private void runTestAndAssertSucceed(String[] testCase) throws Exception {
+ ExecutorService executorService = Executors.newFixedThreadPool(testCase.length);
+ final CountDownLatch countDownLatch = new CountDownLatch(testCase.length);
+ List<Future<List<String>>> tasks = Lists.newArrayListWithExpectedSize(testCase.length);
+ for (int i = 0; i < testCase.length; i++) {
+ tasks.add(executorService.submit(new TestCallable(testCase[i], countDownLatch)));
+ }
+ countDownLatch.await();
+ try {
+ for (int i = 0; i < tasks.size(); ++i) {
+ Future<List<String>> task = tasks.get(i);
+ final List<String> jobIds = task.get();
+ for (String jobId : jobIds) {
+ assertJobSucceed(jobId);
+ }
+ }
+ } catch (Exception ex) {
+ logger.error(ex);
+ throw ex;
+ }
+ }
+
+ private void assertJobSucceed(String jobId) {
+ assertEquals("The job '" + jobId + "' is failed.", ExecutableState.SUCCEED, jobService.getOutput(jobId).getState());
+ }
+
+ private class TestCallable implements Callable<List<String>> {
+
+ private final String methodName;
+ private final CountDownLatch countDownLatch;
+
+ public TestCallable(String methodName, CountDownLatch countDownLatch) {
+ this.methodName = methodName;
+ this.countDownLatch = countDownLatch;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public List<String> call() throws Exception {
+ try {
+ final Method method = BuildCubeWithEngine.class.getDeclaredMethod(methodName);
+ method.setAccessible(true);
+ return (List<String>) method.invoke(BuildCubeWithEngine.this);
+ } catch (Exception e) {
+ logger.error(e.getMessage());
+ throw e;
+ } finally {
+ countDownLatch.countDown();
+ }
+ }
+ }
+
+ @SuppressWarnings("unused")
+ // called by reflection
+ private List<String> testInnerJoinCubeWithSlr() throws Exception {
+ clearSegment("test_kylin_cube_with_slr_empty");
+ SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
+ f.setTimeZone(TimeZone.getTimeZone("GMT"));
+ long date1 = 0;
+ long date2 = f.parse("2015-01-01").getTime();
+ long date3 = f.parse("2022-01-01").getTime();
+ List<String> result = Lists.newArrayList();
+
+ if (fastBuildMode) {
+ result.add(buildSegment("test_kylin_cube_with_slr_empty", date1, date3));
+ } else {
+ result.add(buildSegment("test_kylin_cube_with_slr_empty", date1, date2));
+ result.add(buildSegment("test_kylin_cube_with_slr_empty", date2, date3));//empty segment
+ }
+ return result;
+ }
+
+ @SuppressWarnings("unused")
+ // called by reflection
+ private List<String> testInnerJoinCubeWithoutSlr() throws Exception {
+
+ clearSegment("test_kylin_cube_without_slr_empty");
+ SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
+ f.setTimeZone(TimeZone.getTimeZone("GMT"));
+ long date1 = 0;
+ long date2 = f.parse("2013-01-01").getTime();
+ long date3 = f.parse("2013-07-01").getTime();
+ long date4 = f.parse("2022-01-01").getTime();
+ List<String> result = Lists.newArrayList();
+
+ if (fastBuildMode) {
+ result.add(buildSegment("test_kylin_cube_without_slr_empty", date1, date4));
+ } else {
+ result.add(buildSegment("test_kylin_cube_without_slr_empty", date1, date2));
+ result.add(buildSegment("test_kylin_cube_without_slr_empty", date2, date3));
+ result.add(buildSegment("test_kylin_cube_without_slr_empty", date3, date4));
+ result.add(mergeSegment("test_kylin_cube_without_slr_empty", date1, date3));//don't merge all segments
+ }
+ return result;
+
+ }
+
+ @SuppressWarnings("unused")
+ // called by reflection
+ private List<String> testLeftJoinCubeWithoutSlr() throws Exception {
+ SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
+ f.setTimeZone(TimeZone.getTimeZone("GMT"));
+ List<String> result = Lists.newArrayList();
+ final String cubeName = "test_kylin_cube_without_slr_left_join_empty";
+ clearSegment(cubeName);
+
+ long date1 = cubeManager.getCube(cubeName).getDescriptor().getPartitionDateStart();
+ long date2 = f.parse("2012-06-01").getTime();
+ long date3 = f.parse("2022-01-01").getTime();
+ long date4 = f.parse("2023-01-01").getTime();
+
+ if (fastBuildMode) {
+ result.add(buildSegment("test_kylin_cube_without_slr_left_join_empty", date1, date4));
+ } else {
+ result.add(buildSegment("test_kylin_cube_without_slr_left_join_empty", date1, date2));
+ result.add(buildSegment("test_kylin_cube_without_slr_left_join_empty", date2, date3));
+ result.add(buildSegment("test_kylin_cube_without_slr_left_join_empty", date3, date4));//empty segment
+ result.add(mergeSegment("test_kylin_cube_without_slr_left_join_empty", date1, date3));//don't merge all segments
+ }
+
+ return result;
+
+ }
+
+ @SuppressWarnings("unused")
+ // called by reflection
+ private List<String> testLeftJoinCubeWithSlr() throws Exception {
+ String cubeName = "test_kylin_cube_with_slr_left_join_empty";
+ clearSegment(cubeName);
+
+ SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
+ f.setTimeZone(TimeZone.getTimeZone("GMT"));
+ long date1 = cubeManager.getCube(cubeName).getDescriptor().getPartitionDateStart();
+ long date2 = f.parse("2013-01-01").getTime();
+ long date3 = f.parse("2013-07-01").getTime();
+ long date4 = f.parse("2022-01-01").getTime();
+
+ List<String> result = Lists.newArrayList();
+ if (fastBuildMode) {
+ result.add(buildSegment(cubeName, date1, date4));
+ } else {
+ result.add(buildSegment(cubeName, date1, date2));
+ result.add(buildSegment(cubeName, date2, date3));
+ result.add(buildSegment(cubeName, date3, date4));
+ result.add(mergeSegment(cubeName, date1, date3));//don't merge all segments
+ }
+ return result;
+
+ }
+
+ private void clearSegment(String cubeName) throws Exception {
+ CubeInstance cube = cubeManager.getCube(cubeName);
+ // remove all existing segments
+ CubeUpdate cubeBuilder = new CubeUpdate(cube);
+ cubeBuilder.setToRemoveSegs(cube.getSegments().toArray(new CubeSegment[cube.getSegments().size()]));
+ cubeManager.updateCube(cubeBuilder);
+ }
+
+ private String mergeSegment(String cubeName, long startDate, long endDate) throws Exception {
+ CubeSegment segment = cubeManager.mergeSegments(cubeManager.getCube(cubeName), startDate, endDate, true);
+ DefaultChainedExecutable job = EngineFactory.createBatchMergeJob(segment, "TEST");
+ jobService.addJob(job);
+ waitForJob(job.getId());
+ return job.getId();
+ }
+
+ private String buildSegment(String cubeName, long startDate, long endDate) throws Exception {
+ CubeSegment segment = cubeManager.appendSegments(cubeManager.getCube(cubeName), endDate);
+ DefaultChainedExecutable job = EngineFactory.createBatchCubingJob(segment, "TEST");
+ jobService.addJob(job);
+ waitForJob(job.getId());
+ return job.getId();
+ }
+
+ private int cleanupOldStorage() throws Exception {
+ String[] args = { "--delete", "true" };
+
+ int exitCode = ToolRunner.run(new StorageCleanupJob(), args);
+ return exitCode;
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kylin/blob/7f08dab5/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithSpark.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithSpark.java b/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithSpark.java
new file mode 100644
index 0000000..2a142fc
--- /dev/null
+++ b/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithSpark.java
@@ -0,0 +1,155 @@
+/*
+ * 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.provision;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.File;
+import java.text.SimpleDateFormat;
+import java.util.List;
+import java.util.TimeZone;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.AbstractKylinTestCase;
+import org.apache.kylin.common.util.ClassUtil;
+import org.apache.kylin.common.util.HBaseMetadataTestCase;
+import org.apache.kylin.cube.CubeInstance;
+import org.apache.kylin.cube.CubeManager;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.cube.CubeUpdate;
+import org.apache.kylin.engine.spark.SparkBatchCubingEngine;
+import org.apache.kylin.job.DeployUtil;
+import org.apache.kylin.job.engine.JobEngineConfig;
+import org.apache.kylin.job.execution.AbstractExecutable;
+import org.apache.kylin.job.execution.DefaultChainedExecutable;
+import org.apache.kylin.job.execution.ExecutableState;
+import org.apache.kylin.job.impl.threadpool.DefaultScheduler;
+import org.apache.kylin.job.lock.MockJobLock;
+import org.apache.kylin.job.manager.ExecutableManager;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
+
+//TODO: convert it to a normal class rather than a test case, like in BuildCubeWithEngine
+@Ignore
+public class BuildCubeWithSpark {
+
+ private CubeManager cubeManager;
+ private DefaultScheduler scheduler;
+ protected ExecutableManager jobService;
+
+ private static final Log logger = LogFactory.getLog(BuildCubeWithSpark.class);
+
+ protected void waitForJob(String jobId) {
+ while (true) {
+ AbstractExecutable job = jobService.getJob(jobId);
+ if (job.getStatus() == ExecutableState.SUCCEED || job.getStatus() == ExecutableState.ERROR) {
+ break;
+ } else {
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+ }
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ logger.info("Adding to classpath: " + new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+ ClassUtil.addClasspath(new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+ System.setProperty(KylinConfig.KYLIN_CONF, "../examples/test_case_data/sandbox");
+ if (System.getProperty("hdp.version") == null) {
+ throw new RuntimeException("No hdp.version set; Please set hdp.version in your jvm option, for example: -Dhdp.version=2.2.4.2-2");
+ }
+
+ }
+
+ @Before
+ public void before() throws Exception {
+ HBaseMetadataTestCase.staticCreateTestMetadata(AbstractKylinTestCase.SANDBOX_TEST_DATA);
+
+ DeployUtil.initCliWorkDir();
+ DeployUtil.deployMetadata();
+ DeployUtil.overrideJobJarLocations();
+
+ final KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
+ jobService = ExecutableManager.getInstance(kylinConfig);
+ for (String jobId : jobService.getAllJobIds()) {
+ jobService.deleteJob(jobId);
+ }
+ scheduler = DefaultScheduler.getInstance();
+ scheduler.init(new JobEngineConfig(kylinConfig), new MockJobLock());
+ if (!scheduler.hasStarted()) {
+ throw new RuntimeException("scheduler has not been started");
+ }
+ cubeManager = CubeManager.getInstance(kylinConfig);
+
+ }
+
+ @After
+ public void after() {
+ HBaseMetadataTestCase.staticCleanupTestMetadata();
+ }
+
+ @Test
+ public void test() throws Exception {
+ final CubeSegment segment = createSegment();
+ String confPath = new File(AbstractKylinTestCase.SANDBOX_TEST_DATA).getAbsolutePath();
+ KylinConfig.getInstanceFromEnv().getCoprocessorLocalJar();
+ String coprocessor = KylinConfig.getInstanceFromEnv().getCoprocessorLocalJar();
+ logger.info("confPath location:" + confPath);
+ logger.info("coprocessor location:" + coprocessor);
+ final DefaultChainedExecutable cubingJob = new SparkBatchCubingEngine(confPath, coprocessor).createBatchCubingJob(segment, "BuildCubeWithSpark");
+ jobService.addJob(cubingJob);
+ waitForJob(cubingJob.getId());
+ assertEquals(ExecutableState.SUCCEED, jobService.getOutput(cubingJob.getId()).getState());
+ }
+
+ private void clearSegment(String cubeName) throws Exception {
+ CubeInstance cube = cubeManager.getCube(cubeName);
+ // remove all existing segments
+ CubeUpdate cubeBuilder = new CubeUpdate(cube);
+ cubeBuilder.setToRemoveSegs(cube.getSegments().toArray(new CubeSegment[cube.getSegments().size()]));
+ cubeManager.updateCube(cubeBuilder);
+ }
+
+ private CubeSegment createSegment() throws Exception {
+ String cubeName = "test_kylin_cube_with_slr_left_join_empty";
+ clearSegment(cubeName);
+
+ SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
+ f.setTimeZone(TimeZone.getTimeZone("GMT"));
+ long dateStart = cubeManager.getCube(cubeName).getDescriptor().getPartitionDateStart();
+ long dateEnd = f.parse("2050-11-12").getTime();
+
+ // this cube's start date is 0, end date is 20501112000000
+ List<String> result = Lists.newArrayList();
+ return cubeManager.appendSegments(cubeManager.getCube(cubeName), dateEnd);
+
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kylin/blob/7f08dab5/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithStream.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithStream.java b/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithStream.java
new file mode 100644
index 0000000..290b869
--- /dev/null
+++ b/kylin-it/src/test/java/org/apache/kylin/provision/BuildCubeWithStream.java
@@ -0,0 +1,95 @@
+/*
+ * 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.provision;
+
+import java.io.File;
+import java.util.UUID;
+
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.AbstractKylinTestCase;
+import org.apache.kylin.common.util.ClassUtil;
+import org.apache.kylin.common.util.DateFormat;
+import org.apache.kylin.common.util.HBaseMetadataTestCase;
+import org.apache.kylin.engine.streaming.OneOffStreamingBuilder;
+import org.apache.kylin.engine.streaming.StreamingConfig;
+import org.apache.kylin.engine.streaming.StreamingManager;
+import org.apache.kylin.job.DeployUtil;
+import org.apache.kylin.source.kafka.KafkaConfigManager;
+import org.apache.kylin.source.kafka.config.KafkaConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * for streaming cubing case "test_streaming_table"
+ */
+public class BuildCubeWithStream {
+
+ private static final Logger logger = LoggerFactory.getLogger(BuildCubeWithStream.class);
+ private static final String streamingName = "test_streaming_table_cube";
+ private static final long startTime = DateFormat.stringToMillis("2015-01-01 00:00:00");
+ private static final long endTime = DateFormat.stringToMillis("2015-01-03 00:00:00");
+ private static final long batchInterval = 16 * 60 * 60 * 1000;//16 hours
+
+ private KylinConfig kylinConfig;
+
+ public static void main(String[] args) throws Exception {
+ beforeClass();
+ BuildCubeWithStream buildCubeWithStream = new BuildCubeWithStream();
+ buildCubeWithStream.before();
+ buildCubeWithStream.build();
+ afterClass();
+ }
+
+ public static void beforeClass() throws Exception {
+ logger.info("Adding to classpath: " + new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+ ClassUtil.addClasspath(new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+ System.setProperty(KylinConfig.KYLIN_CONF, "../examples/test_case_data/sandbox");
+ if (System.getProperty("hdp.version") == null) {
+ throw new RuntimeException("No hdp.version set; Please set hdp.version in your jvm option, for example: -Dhdp.version=2.2.4.2-2");
+ }
+ HBaseMetadataTestCase.staticCreateTestMetadata(AbstractKylinTestCase.SANDBOX_TEST_DATA);
+ }
+
+ public void before() throws Exception {
+ DeployUtil.overrideJobJarLocations();
+
+ kylinConfig = KylinConfig.getInstanceFromEnv();
+
+ final StreamingConfig config = StreamingManager.getInstance(KylinConfig.getInstanceFromEnv()).getStreamingConfig(streamingName);
+
+ //Use a random topic for kafka data stream
+ KafkaConfig streamingConfig = KafkaConfigManager.getInstance(kylinConfig).getKafkaConfig(streamingName);
+ streamingConfig.setTopic(UUID.randomUUID().toString());
+ KafkaConfigManager.getInstance(kylinConfig).saveKafkaConfig(streamingConfig);
+
+ DeployUtil.prepareTestDataForStreamingCube(startTime, endTime, config.getCubeName(), streamingConfig);
+ }
+
+ public static void afterClass() throws Exception {
+ HBaseMetadataTestCase.staticCleanupTestMetadata();
+ }
+
+ public void build() throws Exception {
+ logger.info("start time:" + startTime + " end time:" + endTime + " batch interval:" + batchInterval + " batch count:" + ((endTime - startTime) / batchInterval));
+ for (long start = startTime; start < endTime; start += batchInterval) {
+ logger.info(String.format("build batch:{%d, %d}", start, start + batchInterval));
+ new OneOffStreamingBuilder(streamingName, start, start + batchInterval).build().run();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/7f08dab5/kylin-it/src/test/java/org/apache/kylin/provision/BuildIIWithEngine.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/provision/BuildIIWithEngine.java b/kylin-it/src/test/java/org/apache/kylin/provision/BuildIIWithEngine.java
new file mode 100644
index 0000000..b9c242d
--- /dev/null
+++ b/kylin-it/src/test/java/org/apache/kylin/provision/BuildIIWithEngine.java
@@ -0,0 +1,253 @@
+/*
+ * 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.provision;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.File;
+import java.lang.reflect.Method;
+import java.text.SimpleDateFormat;
+import java.util.List;
+import java.util.TimeZone;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.AbstractKylinTestCase;
+import org.apache.kylin.common.util.ClassUtil;
+import org.apache.kylin.common.util.HBaseMetadataTestCase;
+import org.apache.kylin.engine.mr.invertedindex.BatchIIJobBuilder;
+import org.apache.kylin.engine.mr.invertedindex.IIJob;
+import org.apache.kylin.invertedindex.IIInstance;
+import org.apache.kylin.invertedindex.IIManager;
+import org.apache.kylin.invertedindex.IISegment;
+import org.apache.kylin.job.DeployUtil;
+import org.apache.kylin.job.engine.JobEngineConfig;
+import org.apache.kylin.job.execution.AbstractExecutable;
+import org.apache.kylin.job.execution.ExecutableState;
+import org.apache.kylin.job.impl.threadpool.DefaultScheduler;
+import org.apache.kylin.job.manager.ExecutableManager;
+import org.apache.kylin.metadata.realization.RealizationStatusEnum;
+import org.apache.kylin.storage.hbase.util.StorageCleanupJob;
+import org.apache.kylin.storage.hbase.util.ZookeeperJobLock;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
+
+//TODO: convert it to a normal class rather than a test case, like in BuildCubeWithEngine
+@Ignore
+public class BuildIIWithEngine {
+
+ private JobEngineConfig jobEngineConfig;
+ private IIManager iiManager;
+
+ private DefaultScheduler scheduler;
+ protected ExecutableManager jobService;
+
+ protected static final String[] TEST_II_INSTANCES = new String[] { "test_kylin_ii_inner_join", "test_kylin_ii_left_join" };
+
+ private static final Log logger = LogFactory.getLog(BuildIIWithEngine.class);
+
+ protected void waitForJob(String jobId) {
+ while (true) {
+ AbstractExecutable job = jobService.getJob(jobId);
+ if (job.getStatus() == ExecutableState.SUCCEED || job.getStatus() == ExecutableState.ERROR) {
+ break;
+ } else {
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+ }
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ logger.info("Adding to classpath: " + new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+ ClassUtil.addClasspath(new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+ System.setProperty(KylinConfig.KYLIN_CONF, "../examples/test_case_data/sandbox");
+ if (System.getProperty("hdp.version") == null) {
+ throw new RuntimeException("No hdp.version set; Please set hdp.version in your jvm option, for example: -Dhdp.version=2.2.4.2-2");
+ }
+ }
+
+ @Before
+ public void before() throws Exception {
+ HBaseMetadataTestCase.staticCreateTestMetadata(AbstractKylinTestCase.SANDBOX_TEST_DATA);
+
+ //DeployUtil.initCliWorkDir();
+ // DeployUtil.deployMetadata();
+ DeployUtil.overrideJobJarLocations();
+
+ final KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
+ jobService = ExecutableManager.getInstance(kylinConfig);
+ scheduler = DefaultScheduler.getInstance();
+ scheduler.init(new JobEngineConfig(kylinConfig), new ZookeeperJobLock());
+ if (!scheduler.hasStarted()) {
+ throw new RuntimeException("scheduler has not been started");
+ }
+ jobEngineConfig = new JobEngineConfig(kylinConfig);
+ for (String jobId : jobService.getAllJobIds()) {
+ if (jobService.getJob(jobId) instanceof IIJob) {
+ jobService.deleteJob(jobId);
+ }
+ }
+
+ iiManager = IIManager.getInstance(kylinConfig);
+ for (String iiInstance : TEST_II_INSTANCES) {
+
+ IIInstance ii = iiManager.getII(iiInstance);
+ if (ii.getStatus() != RealizationStatusEnum.DISABLED) {
+ ii.setStatus(RealizationStatusEnum.DISABLED);
+ iiManager.updateII(ii);
+ }
+ }
+ }
+
+ @After
+ public void after() throws Exception {
+
+ for (String iiInstance : TEST_II_INSTANCES) {
+ IIInstance ii = iiManager.getII(iiInstance);
+ if (ii.getStatus() != RealizationStatusEnum.READY) {
+ ii.setStatus(RealizationStatusEnum.READY);
+ iiManager.updateII(ii);
+ }
+ }
+ }
+
+ @Test
+ public void testBuildII() throws Exception {
+
+ String[] testCase = new String[] { "buildIIInnerJoin", "buildIILeftJoin" };
+ ExecutorService executorService = Executors.newFixedThreadPool(testCase.length);
+ final CountDownLatch countDownLatch = new CountDownLatch(testCase.length);
+ List<Future<List<String>>> tasks = Lists.newArrayListWithExpectedSize(testCase.length);
+ for (int i = 0; i < testCase.length; i++) {
+ tasks.add(executorService.submit(new TestCallable(testCase[i], countDownLatch)));
+ }
+ countDownLatch.await();
+ for (int i = 0; i < tasks.size(); ++i) {
+ Future<List<String>> task = tasks.get(i);
+ final List<String> jobIds = task.get();
+ for (String jobId : jobIds) {
+ assertJobSucceed(jobId);
+ }
+ }
+
+ }
+
+ private void assertJobSucceed(String jobId) {
+ assertEquals(ExecutableState.SUCCEED, jobService.getOutput(jobId).getState());
+ }
+
+ private class TestCallable implements Callable<List<String>> {
+
+ private final String methodName;
+ private final CountDownLatch countDownLatch;
+
+ public TestCallable(String methodName, CountDownLatch countDownLatch) {
+ this.methodName = methodName;
+ this.countDownLatch = countDownLatch;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public List<String> call() throws Exception {
+ try {
+ final Method method = BuildIIWithEngine.class.getDeclaredMethod(methodName);
+ method.setAccessible(true);
+ return (List<String>) method.invoke(BuildIIWithEngine.this);
+ } finally {
+ countDownLatch.countDown();
+ }
+ }
+ }
+
+ protected List<String> buildIIInnerJoin() throws Exception {
+ return buildII(TEST_II_INSTANCES[0]);
+ }
+
+ protected List<String> buildIILeftJoin() throws Exception {
+ return buildII(TEST_II_INSTANCES[1]);
+ }
+
+ protected List<String> buildII(String iiName) throws Exception {
+ clearSegment(iiName);
+
+ SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
+ f.setTimeZone(TimeZone.getTimeZone("GMT"));
+
+ long date1 = 0;
+ long date2 = f.parse("2015-01-01").getTime();
+
+ List<String> result = Lists.newArrayList();
+ result.add(buildSegment(iiName, date1, date2));
+ return result;
+ }
+
+ private void clearSegment(String iiName) throws Exception {
+ IIInstance ii = iiManager.getII(iiName);
+ ii.getSegments().clear();
+ iiManager.updateII(ii);
+ }
+
+ private String buildSegment(String iiName, long startDate, long endDate) throws Exception {
+ IIInstance iiInstance = iiManager.getII(iiName);
+ IISegment segment = iiManager.buildSegment(iiInstance, startDate, endDate);
+ iiInstance.getSegments().add(segment);
+ iiManager.updateII(iiInstance);
+
+ BatchIIJobBuilder batchIIJobBuilder = new BatchIIJobBuilder(segment, "SYSTEM");
+ IIJob job = batchIIJobBuilder.build();
+ jobService.addJob(job);
+ waitForJob(job.getId());
+ return job.getId();
+ }
+
+ private int cleanupOldStorage() throws Exception {
+ String[] args = { "--delete", "true" };
+
+ int exitCode = ToolRunner.run(new StorageCleanupJob(), args);
+ return exitCode;
+ }
+
+ public static void main(String[] args) throws Exception {
+ BuildIIWithEngine instance = new BuildIIWithEngine();
+
+ BuildIIWithEngine.beforeClass();
+ instance.before();
+ instance.testBuildII();
+ instance.after();
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/7f08dab5/kylin-it/src/test/java/org/apache/kylin/provision/BuildIIWithStream.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/provision/BuildIIWithStream.java b/kylin-it/src/test/java/org/apache/kylin/provision/BuildIIWithStream.java
new file mode 100644
index 0000000..4552cf0
--- /dev/null
+++ b/kylin-it/src/test/java/org/apache/kylin/provision/BuildIIWithStream.java
@@ -0,0 +1,294 @@
+/*
+ *
+ *
+ * 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.provision;
+
+import java.io.File;
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.TimeZone;
+import java.util.UUID;
+
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.AbstractKylinTestCase;
+import org.apache.kylin.common.util.ClassUtil;
+import org.apache.kylin.common.util.DateFormat;
+import org.apache.kylin.common.util.HBaseMetadataTestCase;
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.common.util.StreamingBatch;
+import org.apache.kylin.common.util.StreamingMessage;
+import org.apache.kylin.engine.mr.JobBuilderSupport;
+import org.apache.kylin.invertedindex.IIInstance;
+import org.apache.kylin.invertedindex.IIManager;
+import org.apache.kylin.invertedindex.IISegment;
+import org.apache.kylin.invertedindex.index.Slice;
+import org.apache.kylin.invertedindex.index.SliceBuilder;
+import org.apache.kylin.invertedindex.model.IIDesc;
+import org.apache.kylin.invertedindex.model.IIJoinedFlatTableDesc;
+import org.apache.kylin.invertedindex.model.IIKeyValueCodec;
+import org.apache.kylin.invertedindex.model.IIRow;
+import org.apache.kylin.job.DeployUtil;
+import org.apache.kylin.job.JoinedFlatTable;
+import org.apache.kylin.job.common.ShellExecutable;
+import org.apache.kylin.job.constant.ExecutableConstants;
+import org.apache.kylin.job.engine.JobEngineConfig;
+import org.apache.kylin.metadata.model.TblColRef;
+import org.apache.kylin.metadata.realization.RealizationStatusEnum;
+import org.apache.kylin.source.hive.HiveTableReader;
+import org.apache.kylin.storage.hbase.HBaseConnection;
+import org.apache.kylin.storage.hbase.ii.IICreateHTableJob;
+import org.apache.kylin.storage.hbase.util.StorageCleanupJob;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+public class BuildIIWithStream {
+
+ private static final Logger logger = LoggerFactory.getLogger(BuildIIWithStream.class);
+
+ private static final String[] II_NAME = new String[] { "test_kylin_ii_left_join", "test_kylin_ii_inner_join" };
+ private IIManager iiManager;
+ private KylinConfig kylinConfig;
+
+ public static void main(String[] args) throws Exception {
+ beforeClass();
+ BuildIIWithStream buildCubeWithEngine = new BuildIIWithStream();
+ buildCubeWithEngine.before();
+ buildCubeWithEngine.build();
+ afterClass();
+ }
+
+ public static void beforeClass() throws Exception {
+ logger.info("Adding to classpath: " + new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+ ClassUtil.addClasspath(new File(HBaseMetadataTestCase.SANDBOX_TEST_DATA).getAbsolutePath());
+ if (System.getProperty("hdp.version") == null) {
+ throw new RuntimeException("No hdp.version set; Please set hdp.version in your jvm option, for example: -Dhdp.version=2.2.4.2-2");
+ }
+ HBaseMetadataTestCase.staticCreateTestMetadata(AbstractKylinTestCase.SANDBOX_TEST_DATA);
+ }
+
+ public void before() throws Exception {
+ DeployUtil.overrideJobJarLocations();
+
+ kylinConfig = KylinConfig.getInstanceFromEnv();
+ iiManager = IIManager.getInstance(kylinConfig);
+ for (String iiInstance : II_NAME) {
+
+ IIInstance ii = iiManager.getII(iiInstance);
+ if (ii.getStatus() != RealizationStatusEnum.DISABLED) {
+ ii.setStatus(RealizationStatusEnum.DISABLED);
+ iiManager.updateII(ii);
+ }
+ }
+ }
+
+ public static void afterClass() throws Exception {
+ cleanupOldStorage();
+ HBaseMetadataTestCase.staticCleanupTestMetadata();
+ }
+
+ private String createIntermediateTable(IIDesc desc, KylinConfig kylinConfig) throws IOException {
+ IIJoinedFlatTableDesc intermediateTableDesc = new IIJoinedFlatTableDesc(desc);
+ JobEngineConfig jobEngineConfig = new JobEngineConfig(kylinConfig);
+ final String uuid = UUID.randomUUID().toString();
+ final String useDatabaseHql = "USE " + kylinConfig.getHiveDatabaseForIntermediateTable() + ";";
+ final String dropTableHql = JoinedFlatTable.generateDropTableStatement(intermediateTableDesc);
+ final String createTableHql = JoinedFlatTable.generateCreateTableStatement(intermediateTableDesc, JobBuilderSupport.getJobWorkingDir(jobEngineConfig, uuid));
+ String insertDataHqls;
+ try {
+ insertDataHqls = JoinedFlatTable.generateInsertDataStatement(intermediateTableDesc, jobEngineConfig);
+ } catch (IOException e1) {
+ e1.printStackTrace();
+ throw new RuntimeException("Failed to generate insert data SQL for intermediate table.");
+ }
+
+ ShellExecutable step = new ShellExecutable();
+ StringBuffer buf = new StringBuffer();
+ buf.append("hive -e \"");
+ buf.append(useDatabaseHql + "\n");
+ buf.append(dropTableHql + "\n");
+ buf.append(createTableHql + "\n");
+ buf.append(insertDataHqls + "\n");
+ buf.append("\"");
+
+ step.setCmd(buf.toString());
+ logger.info(step.getCmd());
+ step.setName(ExecutableConstants.STEP_NAME_CREATE_FLAT_HIVE_TABLE);
+ kylinConfig.getCliCommandExecutor().execute(step.getCmd(), null);
+ return intermediateTableDesc.getTableName();
+ }
+
+ private void clearSegment(String iiName) throws Exception {
+ IIInstance ii = iiManager.getII(iiName);
+ ii.getSegments().clear();
+ iiManager.updateII(ii);
+ }
+
+ private IISegment createSegment(String iiName) throws Exception {
+ clearSegment(iiName);
+ SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd");
+ f.setTimeZone(TimeZone.getTimeZone("GMT"));
+
+ long date1 = 0;
+ long date2 = f.parse("2015-01-01").getTime();
+ return buildSegment(iiName, date1, date2);
+ }
+
+ private IISegment buildSegment(String iiName, long startDate, long endDate) throws Exception {
+ IIInstance iiInstance = iiManager.getII(iiName);
+ IISegment segment = iiManager.buildSegment(iiInstance, startDate, endDate);
+ iiInstance.getSegments().add(segment);
+ iiManager.updateII(iiInstance);
+ return segment;
+ }
+
+ private void buildII(String iiName) throws Exception {
+ final IIDesc desc = iiManager.getII(iiName).getDescriptor();
+ final String tableName = createIntermediateTable(desc, kylinConfig);
+ logger.info("intermediate table name:" + tableName);
+
+ HiveTableReader reader = new HiveTableReader("default", tableName);
+ final List<TblColRef> tblColRefs = desc.listAllColumns();
+ for (TblColRef tblColRef : tblColRefs) {
+ if (desc.isMetricsCol(tblColRef)) {
+ logger.info("matrix:" + tblColRef.getName());
+ } else {
+ logger.info("measure:" + tblColRef.getName());
+ }
+ }
+ final IISegment segment = createSegment(iiName);
+ final HTableInterface htable = HBaseConnection.get(KylinConfig.getInstanceFromEnv().getStorageUrl()).getTable(segment.getStorageLocationIdentifier());
+ String[] args = new String[] { "-iiname", iiName, "-htablename", segment.getStorageLocationIdentifier() };
+ ToolRunner.run(new IICreateHTableJob(), args);
+
+ final IIDesc iiDesc = segment.getIIDesc();
+ final SliceBuilder sliceBuilder = new SliceBuilder(desc, (short) 0);
+
+ List<String[]> sorted = getSortedRows(reader, desc.getTimestampColumn());
+ int count = sorted.size();
+ ArrayList<StreamingMessage> messages = Lists.newArrayList();
+ for (String[] row : sorted) {
+ messages.add((parse(row)));
+ if (messages.size() >= iiDesc.getSliceSize()) {
+ build(sliceBuilder, new StreamingBatch(messages, Pair.newPair(System.currentTimeMillis(), System.currentTimeMillis())), htable);
+ messages.clear();
+ }
+ }
+
+ if (!messages.isEmpty()) {
+ build(sliceBuilder, new StreamingBatch(messages, Pair.newPair(System.currentTimeMillis(), System.currentTimeMillis())), htable);
+ }
+
+ reader.close();
+ logger.info("total record count:" + count + " htable:" + segment.getStorageLocationIdentifier());
+ logger.info("stream build finished, htable name:" + segment.getStorageLocationIdentifier());
+ }
+
+ public void build() throws Exception {
+ for (String iiName : II_NAME) {
+ buildII(iiName);
+ IIInstance ii = iiManager.getII(iiName);
+ if (ii.getStatus() != RealizationStatusEnum.READY) {
+ ii.setStatus(RealizationStatusEnum.READY);
+ iiManager.updateII(ii);
+ }
+ }
+ }
+
+ private void build(SliceBuilder sliceBuilder, StreamingBatch batch, HTableInterface htable) throws IOException {
+ final Slice slice = sliceBuilder.buildSlice(batch);
+ try {
+ loadToHBase(htable, slice, new IIKeyValueCodec(slice.getInfo()));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void loadToHBase(HTableInterface hTable, Slice slice, IIKeyValueCodec codec) throws IOException {
+ List<Put> data = Lists.newArrayList();
+ for (IIRow row : codec.encodeKeyValue(slice)) {
+ final byte[] key = row.getKey().get();
+ final byte[] value = row.getValue().get();
+ Put put = new Put(key);
+ put.add(IIDesc.HBASE_FAMILY_BYTES, IIDesc.HBASE_QUALIFIER_BYTES, value);
+ final ImmutableBytesWritable dictionary = row.getDictionary();
+ final byte[] dictBytes = dictionary.get();
+ if (dictionary.getOffset() == 0 && dictionary.getLength() == dictBytes.length) {
+ put.add(IIDesc.HBASE_FAMILY_BYTES, IIDesc.HBASE_DICTIONARY_BYTES, dictBytes);
+ } else {
+ throw new RuntimeException("dict offset should be 0, and dict length should be " + dictBytes.length + " but they are" + dictionary.getOffset() + " " + dictionary.getLength());
+ }
+ data.add(put);
+ }
+ hTable.put(data);
+ //omit hTable.flushCommits(), because htable is auto flush
+ }
+
+ private StreamingMessage parse(String[] row) {
+ return new StreamingMessage(Lists.newArrayList(row), System.currentTimeMillis(), System.currentTimeMillis(), Collections.<String, Object> emptyMap());
+ }
+
+ private List<String[]> getSortedRows(HiveTableReader reader, final int tsCol) throws IOException {
+ List<String[]> unsorted = Lists.newArrayList();
+ while (reader.next()) {
+ unsorted.add(reader.getRow());
+ }
+ Collections.sort(unsorted, new Comparator<String[]>() {
+ @Override
+ public int compare(String[] o1, String[] o2) {
+ long t1 = DateFormat.stringToMillis(o1[tsCol]);
+ long t2 = DateFormat.stringToMillis(o2[tsCol]);
+ return Long.compare(t1, t2);
+ }
+ });
+ return unsorted;
+ }
+
+ private static int cleanupOldStorage() throws Exception {
+ String[] args = { "--delete", "true" };
+
+ int exitCode = ToolRunner.run(new StorageCleanupJob(), args);
+ return exitCode;
+ }
+
+}