Repository: carbondata Updated Branches: refs/heads/carbonstore 7306b59dd -> 0aab4e7c6
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0aab4e7c/pom.xml ---------------------------------------------------------------------- diff --git a/pom.xml b/pom.xml index d6436d7..74e5c01 100644 --- a/pom.xml +++ b/pom.xml @@ -560,7 +560,6 @@ <sourceDirectory>${basedir}/hadoop/src/main/java</sourceDirectory> <sourceDirectory>${basedir}/integration/spark2/src/main/scala</sourceDirectory> <sourceDirectory>${basedir}/integration/spark2/src/main/spark2.2</sourceDirectory> - <sourceDirectory>${basedir}/integration/spark2/src/main/commonTo2.2And2.3</sourceDirectory> <sourceDirectory>${basedir}/integration/spark2/src/main/java</sourceDirectory> <sourceDirectory>${basedir}/integration/spark-common/src/main/scala</sourceDirectory> <sourceDirectory>${basedir}/integration/spark-common/src/main/java</sourceDirectory> @@ -591,60 +590,6 @@ </modules> </profile> <profile> - <id>spark-2.3</id> - <properties> - <spark.version>2.3.1</spark.version> - <scala.binary.version>2.11</scala.binary.version> - <scala.version>2.11.8</scala.version> - </properties> - <modules> - <module>integration/spark2</module> - <module>integration/hive</module> - <module>integration/presto</module> - <module>streaming</module> - <module>examples/spark2</module> - <module>datamap/lucene</module> - <module>datamap/bloom</module> - </modules> - <build> - <plugins> - <plugin> - <groupId>org.eluder.coveralls</groupId> - <artifactId>coveralls-maven-plugin</artifactId> - <version>4.3.0</version> - <configuration> - <repoToken>opPwqWW41vYppv6KISea3u1TJvE1ugJ5Y</repoToken> - <sourceEncoding>UTF-8</sourceEncoding> - <jacocoReports> - <jacocoReport>${basedir}/target/carbondata-coverage-report/carbondata-coverage-report.xml - </jacocoReport> - </jacocoReports> - <sourceDirectories> - <sourceDirectory>${basedir}/common/src/main/java</sourceDirectory> - <sourceDirectory>${basedir}/core/src/main/java</sourceDirectory> - <sourceDirectory>${basedir}/processing/src/main/java</sourceDirectory> - <sourceDirectory>${basedir}/hadoop/src/main/java</sourceDirectory> - <sourceDirectory>${basedir}/integration/spark2/src/main/scala</sourceDirectory> - <sourceDirectory>${basedir}/integration/spark2/src/main/commonTo2.2And2.3</sourceDirectory> - <sourceDirectory>${basedir}/integration/spark2/src/main/spark2.3</sourceDirectory> - <sourceDirectory>${basedir}/integration/spark2/src/main/java</sourceDirectory> - <sourceDirectory>${basedir}/integration/spark-common/src/main/scala</sourceDirectory> - <sourceDirectory>${basedir}/integration/spark-common/src/main/java</sourceDirectory> - <sourceDirectory>${basedir}/integration/spark-common-test/src/main/scala</sourceDirectory> - <sourceDirectory>${basedir}/integration/spark-common-test/src/main/java</sourceDirectory> - <sourceDirectory>${basedir}/integration/hive/src/main/scala</sourceDirectory> - <sourceDirectory>${basedir}/integration/hive/src/main/java</sourceDirectory> - <sourceDirectory>${basedir}/integration/presto/src/main/scala</sourceDirectory> - <sourceDirectory>${basedir}/integration/presto/src/main/java</sourceDirectory> - <sourceDirectory>${basedir}/streaming/src/main/java</sourceDirectory> - <sourceDirectory>${basedir}/streaming/src/main/scala</sourceDirectory> - </sourceDirectories> - </configuration> - </plugin> - </plugins> - </build> - </profile> - <profile> <id>include-all</id> </profile> <profile> http://git-wip-us.apache.org/repos/asf/carbondata/blob/0aab4e7c/streaming/src/test/java/org/apache/carbondata/streaming/CarbonStreamInputFormatTest.java ---------------------------------------------------------------------- diff --git a/streaming/src/test/java/org/apache/carbondata/streaming/CarbonStreamInputFormatTest.java b/streaming/src/test/java/org/apache/carbondata/streaming/CarbonStreamInputFormatTest.java new file mode 100644 index 0000000..a224446 --- /dev/null +++ b/streaming/src/test/java/org/apache/carbondata/streaming/CarbonStreamInputFormatTest.java @@ -0,0 +1,99 @@ +/* + * 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.carbondata.streaming; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Date; +import java.util.List; +import java.util.UUID; + +import org.apache.carbondata.core.datastore.impl.FileFactory; +import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier; +import org.apache.carbondata.core.metadata.CarbonTableIdentifier; +import org.apache.carbondata.core.statusmanager.FileFormat; +import org.apache.carbondata.hadoop.CarbonInputSplit; +import org.apache.carbondata.hadoop.CarbonMultiBlockSplit; +import org.apache.carbondata.hadoop.util.CarbonInputFormatUtil; + +import junit.framework.TestCase; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.mapreduce.InputSplit; +import org.apache.hadoop.mapreduce.JobID; +import org.apache.hadoop.mapreduce.RecordReader; +import org.apache.hadoop.mapreduce.TaskAttemptContext; +import org.apache.hadoop.mapreduce.TaskAttemptID; +import org.apache.hadoop.mapreduce.TaskID; +import org.apache.hadoop.mapreduce.TaskType; +import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl; +import org.junit.Assert; +import org.junit.Test; + +public class CarbonStreamInputFormatTest extends TestCase { + + private TaskAttemptID taskAttemptId; + private TaskAttemptContext taskAttemptContext; + private Configuration hadoopConf; + private AbsoluteTableIdentifier identifier; + private String tablePath; + + + @Override protected void setUp() throws Exception { + tablePath = new File("target/stream_input").getCanonicalPath(); + String dbName = "default"; + String tableName = "stream_table_input"; + identifier = AbsoluteTableIdentifier.from( + tablePath, + new CarbonTableIdentifier(dbName, tableName, UUID.randomUUID().toString())); + + JobID jobId = CarbonInputFormatUtil.getJobId(new Date(), 0); + TaskID taskId = new TaskID(jobId, TaskType.MAP, 0); + taskAttemptId = new TaskAttemptID(taskId, 0); + + hadoopConf = new Configuration(); + taskAttemptContext = new TaskAttemptContextImpl(hadoopConf, taskAttemptId); + } + + private InputSplit buildInputSplit() throws IOException { + CarbonInputSplit carbonInputSplit = new CarbonInputSplit(); + List<CarbonInputSplit> splitList = new ArrayList<>(); + splitList.add(carbonInputSplit); + return new CarbonMultiBlockSplit(splitList, new String[] { "localhost" }, + FileFormat.ROW_V1); + } + + @Test public void testCreateRecordReader() { + try { + InputSplit inputSplit = buildInputSplit(); + CarbonStreamInputFormat inputFormat = new CarbonStreamInputFormat(); + RecordReader recordReader = inputFormat.createRecordReader(inputSplit, taskAttemptContext); + Assert.assertNotNull("Failed to create record reader", recordReader); + } catch (Exception e) { + e.printStackTrace(); + Assert.assertTrue(e.getMessage(), false); + } + } + + @Override protected void tearDown() throws Exception { + super.tearDown(); + if (tablePath != null) { + FileFactory.deleteAllFilesOfDir(new File(tablePath)); + } + } +}
