| /* |
| * 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. |
| */ |
| |
| #include <iostream> |
| #include <jni.h> |
| #include <stdio.h> |
| #include <stdlib.h> |
| #include <string.h> |
| #include <sys/time.h> |
| #include <unistd.h> |
| #include <sys/time.h> |
| #include "../src/CarbonReader.h" |
| #include "../src/CarbonRow.h" |
| #include "../src/CarbonWriter.h" |
| #include "../src/CarbonSchemaReader.h" |
| #include "../src/Schema.h" |
| #include "../src/CarbonProperties.h" |
| |
| using namespace std; |
| |
| JavaVM *jvm; |
| |
| /** |
| * init jvm |
| * |
| * @return |
| */ |
| JNIEnv *initJVM() { |
| JNIEnv *env; |
| JavaVMInitArgs vm_args; |
| int parNum = 2; |
| int res; |
| JavaVMOption options[parNum]; |
| |
| options[0].optionString = "-Djava.class.path=../../sdk/target/carbondata-sdk.jar"; |
| options[1].optionString = "-verbose:jni"; // For debug and check the jni information |
| // options[2].optionString = "-Xmx12000m"; // change the jvm max memory size |
| // options[3].optionString = "-Xms5000m"; // change the jvm min memory size |
| // options[4].optionString = "-Djava.compiler=NONE"; // forbidden JIT |
| vm_args.version = JNI_VERSION_1_8; |
| vm_args.nOptions = parNum; |
| vm_args.options = options; |
| vm_args.ignoreUnrecognized = JNI_FALSE; |
| |
| res = JNI_CreateJavaVM(&jvm, (void **) &env, &vm_args); |
| if (res < 0) { |
| fprintf(stderr, "\nCan't create Java VM\n"); |
| exit(1); |
| } |
| |
| return env; |
| } |
| |
| /** |
| * print array result |
| * |
| * @param env JNIEnv |
| * @param arr array |
| */ |
| void printArray(JNIEnv *env, jobjectArray arr) { |
| if (env->ExceptionCheck()) { |
| throw env->ExceptionOccurred(); |
| } |
| jsize length = env->GetArrayLength(arr); |
| int j = 0; |
| for (j = 0; j < length; j++) { |
| jobject element = env->GetObjectArrayElement(arr, j); |
| if (env->ExceptionCheck()) { |
| throw env->ExceptionOccurred(); |
| } |
| char *str = (char *) env->GetStringUTFChars((jstring) element, JNI_FALSE); |
| printf("%s\t", str); |
| } |
| env->DeleteLocalRef(arr); |
| } |
| |
| /** |
| * print boolean result |
| * |
| * @param env JNIEnv |
| * @param bool1 boolean value |
| */ |
| void printBoolean(jboolean bool1) { |
| if (bool1) { |
| printf("true\t"); |
| } else { |
| printf("false\t"); |
| } |
| } |
| |
| /** |
| * print result of reading data |
| * |
| * @param env JNIEnv |
| * @param reader CarbonReader object |
| */ |
| void printResultWithException(JNIEnv *env, CarbonReader reader) { |
| try { |
| CarbonRow carbonRow(env); |
| while (reader.hasNext()) { |
| jobject row = reader.readNextRow(); |
| carbonRow.setCarbonRow(row); |
| printf("%s\t", carbonRow.getString(1)); |
| printf("%d\t", carbonRow.getInt(1)); |
| printf("%ld\t", carbonRow.getLong(2)); |
| printf("%s\t", carbonRow.getVarchar(1)); |
| printArray(env, carbonRow.getArray(0)); |
| printf("%d\t", carbonRow.getShort(5)); |
| printf("%d\t", carbonRow.getInt(6)); |
| printf("%ld\t", carbonRow.getLong(7)); |
| printf("%lf\t", carbonRow.getDouble(8)); |
| printBoolean(carbonRow.getBoolean(9)); |
| printf("%s\t", carbonRow.getDecimal(9)); |
| printf("%f\t", carbonRow.getFloat(11)); |
| printf("\n"); |
| env->DeleteLocalRef(row); |
| } |
| reader.close(); |
| carbonRow.close(); |
| } catch (jthrowable e) { |
| env->ExceptionDescribe(); |
| } |
| } |
| |
| /** |
| * print result of reading data |
| * |
| * @param env JNIEnv |
| * @param reader CarbonReader object |
| */ |
| void printResult(JNIEnv *env, CarbonReader reader) { |
| try { |
| CarbonRow carbonRow(env); |
| while (reader.hasNext()) { |
| jobject row = reader.readNextRow(); |
| carbonRow.setCarbonRow(row); |
| printf("%s\t", carbonRow.getString(0)); |
| printf("%d\t", carbonRow.getInt(1)); |
| printf("%ld\t", carbonRow.getLong(2)); |
| printf("%s\t", carbonRow.getVarchar(3)); |
| printArray(env, carbonRow.getArray(4)); |
| printf("%d\t", carbonRow.getShort(5)); |
| printf("%d\t", carbonRow.getInt(6)); |
| printf("%ld\t", carbonRow.getLong(7)); |
| printf("%lf\t", carbonRow.getDouble(8)); |
| printBoolean(carbonRow.getBoolean(9)); |
| printf("%s\t", carbonRow.getDecimal(10)); |
| printf("%f\t", carbonRow.getFloat(11)); |
| printf("\n"); |
| env->DeleteLocalRef(row); |
| } |
| carbonRow.close(); |
| reader.close(); |
| } catch (jthrowable e) { |
| env->ExceptionDescribe(); |
| } |
| } |
| |
| /** |
| * test read Schema from path |
| * |
| * @param env jni env |
| * @return whether it is success |
| */ |
| bool readSchema(JNIEnv *env, char *Path, bool validateSchema, char **argv, int argc) { |
| try { |
| printf("\nread Schema:\n"); |
| Configuration conf(env); |
| if (argc > 3) { |
| conf.set("fs.s3a.access.key", argv[1]); |
| conf.set("fs.s3a.secret.key", argv[2]); |
| conf.set("fs.s3a.endpoint", argv[3]); |
| } |
| printf("%s\n", conf.get("fs.s3a.endpoint", "default")); |
| |
| CarbonSchemaReader carbonSchemaReader(env); |
| jobject schema; |
| |
| if (validateSchema) { |
| schema = carbonSchemaReader.readSchema(Path, validateSchema, conf); |
| } else { |
| schema = carbonSchemaReader.readSchema(Path, conf); |
| } |
| Schema carbonSchema(env, schema); |
| int length = carbonSchema.getFieldsLength(); |
| printf("schema length is:%d\n", length); |
| for (int i = 0; i < length; i++) { |
| printf("%d\t", i); |
| printf("%s\t", carbonSchema.getFieldName(i)); |
| printf("%s\n", carbonSchema.getFieldDataTypeName(i)); |
| if (strcmp(carbonSchema.getFieldDataTypeName(i), "ARRAY") == 0) { |
| printf("Array Element Type Name is:%s\n", carbonSchema.getArrayElementTypeName(i)); |
| } |
| } |
| } catch (jthrowable e) { |
| env->ExceptionDescribe(); |
| } |
| return true; |
| } |
| |
| /** |
| * test the exception when carbonRow with wrong index. |
| * |
| * @param env jni env |
| * @return |
| */ |
| bool tryCarbonRowException(JNIEnv *env, char *path) { |
| printf("\nRead data from local without projection:\n"); |
| |
| CarbonReader carbonReaderClass; |
| try { |
| carbonReaderClass.builder(env, path); |
| } catch (runtime_error e) { |
| printf("\nget exception fro builder and throw\n"); |
| throw e; |
| } |
| try { |
| carbonReaderClass.build(); |
| } catch (jthrowable e) { |
| env->ExceptionDescribe(); |
| } |
| printResultWithException(env, carbonReaderClass); |
| } |
| |
| /** |
| * test read data from local disk, without projection |
| * |
| * @param env jni env |
| * @return |
| */ |
| bool readFromLocalWithoutProjection(JNIEnv *env, char *path) { |
| printf("\nRead data from local without projection:\n"); |
| |
| CarbonReader carbonReaderClass; |
| try { |
| carbonReaderClass.builder(env, path); |
| } catch (runtime_error e) { |
| printf("\nget exception fro builder and throw\n"); |
| throw e; |
| } |
| try { |
| carbonReaderClass.build(); |
| } catch (jthrowable e) { |
| env->ExceptionDescribe(); |
| env->ExceptionClear(); |
| } |
| printResult(env, carbonReaderClass); |
| } |
| |
| /** |
| * test read data by readNextRow method |
| * |
| * @param env jni env |
| */ |
| void testReadNextRow(JNIEnv *env, char *path, int printNum, char **argv, int argc, bool useVectorReader) { |
| printf("\nTest next Row Performance, useVectorReader is "); |
| printBoolean(useVectorReader); |
| printf("\n"); |
| |
| try { |
| struct timeval start, build, startRead, endBatchRead, endRead; |
| gettimeofday(&start, NULL); |
| CarbonReader carbonReaderClass; |
| |
| carbonReaderClass.builder(env, path); |
| if (argc > 1) { |
| carbonReaderClass.withHadoopConf("fs.s3a.access.key", argv[1]); |
| carbonReaderClass.withHadoopConf("fs.s3a.secret.key", argv[2]); |
| carbonReaderClass.withHadoopConf("fs.s3a.endpoint", argv[3]); |
| } |
| if (!useVectorReader) { |
| carbonReaderClass.withRowRecordReader(); |
| } |
| carbonReaderClass.build(); |
| |
| gettimeofday(&build, NULL); |
| int time = 1000000 * (build.tv_sec - start.tv_sec) + build.tv_usec - start.tv_usec; |
| double buildTime = time / 1000000.0; |
| printf("\n\nbuild time is: %lf s\n\n", time / 1000000.0); |
| |
| CarbonRow carbonRow(env); |
| int i = 0; |
| |
| gettimeofday(&startRead, NULL); |
| jobject row; |
| while (carbonReaderClass.hasNext()) { |
| |
| row = carbonReaderClass.readNextRow(); |
| |
| i++; |
| if (i > 1 && i % printNum == 0) { |
| gettimeofday(&endBatchRead, NULL); |
| |
| time = 1000000 * (endBatchRead.tv_sec - startRead.tv_sec) + endBatchRead.tv_usec - startRead.tv_usec; |
| printf("%d: time is %lf s, speed is %lf records/s ", i, time / 1000000.0, |
| printNum / (time / 1000000.0)); |
| |
| carbonRow.setCarbonRow(row); |
| printf("%s\t", carbonRow.getString(0)); |
| printf("%s\t", carbonRow.getString(1)); |
| printf("%s\t", carbonRow.getString(2)); |
| printf("%s\t", carbonRow.getString(3)); |
| printf("%ld\t", carbonRow.getLong(4)); |
| printf("%ld\t", carbonRow.getLong(5)); |
| printf("\n"); |
| |
| gettimeofday(&startRead, NULL); |
| } |
| env->DeleteLocalRef(row); |
| } |
| |
| gettimeofday(&endRead, NULL); |
| |
| time = 1000000 * (endRead.tv_sec - build.tv_sec) + endRead.tv_usec - build.tv_usec; |
| printf("total line is: %d,\t build time is: %lf s,\tread time is %lf s, average speed is %lf records/s ", |
| i, buildTime, time / 1000000.0, i / (time / 1000000.0)); |
| carbonReaderClass.close(); |
| carbonRow.close(); |
| } catch (jthrowable) { |
| env->ExceptionDescribe(); |
| env->ExceptionClear(); |
| } |
| } |
| |
| /** |
| * test read data by readNextBatchRow method |
| * |
| * @param env jni env |
| */ |
| void testReadNextBatchRow(JNIEnv *env, char *path, int batchSize, int printNum, char **argv, int argc, |
| bool useVectorReader) { |
| try { |
| printf("\n\nTest next Batch Row Performance:\n"); |
| printBoolean(useVectorReader); |
| printf("\n"); |
| |
| struct timeval start, build, read; |
| gettimeofday(&start, NULL); |
| |
| CarbonReader carbonReaderClass; |
| |
| carbonReaderClass.builder(env, path); |
| if (argc > 1) { |
| carbonReaderClass.withHadoopConf("fs.s3a.access.key", argv[1]); |
| carbonReaderClass.withHadoopConf("fs.s3a.secret.key", argv[2]); |
| carbonReaderClass.withHadoopConf("fs.s3a.endpoint", argv[3]); |
| } |
| if (!useVectorReader) { |
| carbonReaderClass.withRowRecordReader(); |
| } |
| carbonReaderClass.withBatch(batchSize); |
| carbonReaderClass.build(); |
| |
| gettimeofday(&build, NULL); |
| int time = 1000000 * (build.tv_sec - start.tv_sec) + build.tv_usec - start.tv_usec; |
| double buildTime = time / 1000000.0; |
| printf("\n\nbuild time is: %lf s\n\n", time / 1000000.0); |
| |
| CarbonRow carbonRow(env); |
| int i = 0; |
| struct timeval startHasNext, startReadNextBatchRow, endReadNextBatchRow, endRead; |
| gettimeofday(&startHasNext, NULL); |
| |
| while (carbonReaderClass.hasNext()) { |
| |
| gettimeofday(&startReadNextBatchRow, NULL); |
| jobjectArray batch = carbonReaderClass.readNextBatchRow(); |
| if (env->ExceptionCheck()) { |
| env->ExceptionDescribe(); |
| } |
| gettimeofday(&endReadNextBatchRow, NULL); |
| |
| jsize length = env->GetArrayLength(batch); |
| if (i + length > printNum - 1) { |
| for (int j = 0; j < length; j++) { |
| i++; |
| jobject row = env->GetObjectArrayElement(batch, j); |
| carbonRow.setCarbonRow(row); |
| carbonRow.getString(0); |
| carbonRow.getString(1); |
| carbonRow.getString(2); |
| carbonRow.getString(3); |
| carbonRow.getLong(4); |
| carbonRow.getLong(5); |
| if (i > 1 && i % printNum == 0) { |
| gettimeofday(&read, NULL); |
| |
| double hasNextTime = 1000000 * (startReadNextBatchRow.tv_sec - startHasNext.tv_sec) + |
| startReadNextBatchRow.tv_usec - startHasNext.tv_usec; |
| |
| double readNextBatchTime = |
| 1000000 * (endReadNextBatchRow.tv_sec - startReadNextBatchRow.tv_sec) + |
| endReadNextBatchRow.tv_usec - startReadNextBatchRow.tv_usec; |
| |
| time = 1000000 * (read.tv_sec - startHasNext.tv_sec) + read.tv_usec - startHasNext.tv_usec; |
| printf("%d: time is %lf s, speed is %lf records/s, hasNext time is %lf s,readNextBatchRow time is %lf s ", |
| i, time / 1000000.0, printNum / (time / 1000000.0), hasNextTime / 1000000.0, |
| readNextBatchTime / 1000000.0); |
| gettimeofday(&startHasNext, NULL); |
| printf("%s\t", carbonRow.getString(0)); |
| printf("%s\t", carbonRow.getString(1)); |
| printf("%s\t", carbonRow.getString(2)); |
| printf("%s\t", carbonRow.getString(3)); |
| printf("%ld\t", carbonRow.getLong(4)); |
| printf("%ld\t", carbonRow.getLong(5)); |
| printf("\n"); |
| } |
| env->DeleteLocalRef(row); |
| } |
| } else { |
| i = i + length; |
| } |
| env->DeleteLocalRef(batch); |
| } |
| gettimeofday(&endRead, NULL); |
| time = 1000000 * (endRead.tv_sec - build.tv_sec) + endRead.tv_usec - build.tv_usec; |
| printf("total line is: %d,\t build time is: %lf s,\tread time is %lf s, average speed is %lf records/s ", |
| i, buildTime, time / 1000000.0, i / (time / 1000000.0)); |
| carbonReaderClass.close(); |
| carbonRow.close(); |
| } catch (jthrowable e) { |
| env->ExceptionDescribe(); |
| env->ExceptionClear(); |
| } |
| } |
| |
| /** |
| * test read data from local disk |
| * |
| * @param env jni env |
| * @return |
| */ |
| bool readFromLocalWithProjection(JNIEnv *env, char *path) { |
| printf("\nRead data from local:\n"); |
| try { |
| CarbonReader reader; |
| reader.builder(env, path, "test"); |
| |
| char *argv[12]; |
| argv[0] = "stringField"; |
| argv[1] = "shortField"; |
| argv[2] = "intField"; |
| argv[3] = "longField"; |
| argv[4] = "doubleField"; |
| argv[5] = "boolField"; |
| argv[6] = "dateField"; |
| argv[7] = "timeField"; |
| argv[8] = "decimalField"; |
| argv[9] = "varcharField"; |
| argv[10] = "arrayField"; |
| argv[11] = "floatField"; |
| reader.projection(12, argv); |
| |
| reader.build(); |
| |
| CarbonRow carbonRow(env); |
| while (reader.hasNext()) { |
| jobject row = reader.readNextRow(); |
| carbonRow.setCarbonRow(row); |
| |
| printf("%s\t", carbonRow.getString(0)); |
| printf("%d\t", carbonRow.getShort(1)); |
| printf("%d\t", carbonRow.getInt(2)); |
| printf("%ld\t", carbonRow.getLong(3)); |
| printf("%lf\t", carbonRow.getDouble(4)); |
| printBoolean(carbonRow.getBoolean(5)); |
| printf("%d\t", carbonRow.getInt(6)); |
| printf("%ld\t", carbonRow.getLong(7)); |
| printf("%s\t", carbonRow.getDecimal(8)); |
| printf("%s\t", carbonRow.getVarchar(9)); |
| printArray(env, carbonRow.getArray(10)); |
| printf("%f\t", carbonRow.getFloat(11)); |
| printf("\n"); |
| env->DeleteLocalRef(row); |
| } |
| |
| reader.close(); |
| carbonRow.close(); |
| } catch (jthrowable e) { |
| env->ExceptionDescribe(); |
| env->ExceptionClear(); |
| } |
| } |
| |
| |
| bool tryCatchException(JNIEnv *env) { |
| printf("\ntry catch exception and print:\n"); |
| |
| CarbonReader carbonReaderClass; |
| carbonReaderClass.builder(env, "./carbondata"); |
| try { |
| carbonReaderClass.build(); |
| } catch (jthrowable e) { |
| env->ExceptionDescribe(); |
| env->ExceptionClear(); |
| } |
| printf("\nfinished handle exception\n"); |
| } |
| |
| void testCarbonProperties(JNIEnv *env) { |
| try { |
| printf("%s", "test Carbon Properties:"); |
| CarbonProperties carbonProperties(env); |
| char *key = "carbon.unsafe.working.memory.in.mb"; |
| printf("%s\t", carbonProperties.getProperty(key)); |
| printf("%s\t", carbonProperties.getProperty(key, "512")); |
| carbonProperties.addProperty(key, "1024"); |
| printf("%s\t", carbonProperties.getProperty(key)); |
| } catch (jthrowable e) { |
| env->ExceptionDescribe(); |
| env->ExceptionClear(); |
| } |
| } |
| |
| bool testValidateBadRecordsActionWithImproperValue(JNIEnv *env, char *path) { |
| char *jsonSchema = "[{stringField:string},{shortField:short},{intField:int},{longField:long},{doubleField:double},{boolField:boolean},{dateField:date},{timeField:timestamp},{floatField:float},{arrayField:array}]"; |
| try { |
| CarbonWriter writer; |
| writer.builder(env); |
| writer.outputPath(path); |
| writer.withCsvInput(jsonSchema); |
| writer.withLoadOption("BAD_RECORDS_ACTION", "FAL"); |
| writer.writtenBy("CSDK"); |
| writer.build(); |
| } catch (jthrowable ex) { |
| env->ExceptionDescribe(); |
| env->ExceptionClear(); |
| } |
| } |
| |
| bool testValidateBadRecordsLoggerEnableWithImproperValue(JNIEnv *env, char *path) { |
| char *jsonSchema = "[{stringField:string},{shortField:short},{intField:int},{longField:long},{doubleField:double},{boolField:boolean},{dateField:date},{timeField:timestamp},{floatField:float},{arrayField:array}]"; |
| try { |
| CarbonWriter writer; |
| writer.builder(env); |
| writer.outputPath(path); |
| writer.withCsvInput(jsonSchema); |
| writer.withLoadOption("bad_records_logger_enable", "FLSE"); |
| writer.writtenBy("CSDK"); |
| writer.build(); |
| } catch (jthrowable ex) { |
| env->ExceptionDescribe(); |
| env->ExceptionClear(); |
| } |
| } |
| |
| bool testValidateQuoteCharWithImproperValue(JNIEnv *env, char *path) { |
| char *jsonSchema = "[{stringField:string},{shortField:short},{intField:int},{longField:long},{doubleField:double},{boolField:boolean},{dateField:date},{timeField:timestamp},{floatField:float},{arrayField:array}]"; |
| try { |
| CarbonWriter writer; |
| writer.builder(env); |
| writer.outputPath(path); |
| writer.withCsvInput(jsonSchema); |
| writer.withLoadOption("quotechar", "##"); |
| writer.writtenBy("CSDK"); |
| writer.build(); |
| } catch (jthrowable ex) { |
| env->ExceptionDescribe(); |
| env->ExceptionClear(); |
| } |
| } |
| |
| bool testValidateEscapeCharWithImproperValue(JNIEnv *env, char *path) { |
| char *jsonSchema = "[{stringField:string},{shortField:short},{intField:int},{longField:long},{doubleField:double},{boolField:boolean},{dateField:date},{timeField:timestamp},{floatField:float},{arrayField:array}]"; |
| try { |
| CarbonWriter writer; |
| writer.builder(env); |
| writer.outputPath(path); |
| writer.withCsvInput(jsonSchema); |
| writer.withLoadOption("escapechar", "##"); |
| writer.writtenBy("CSDK"); |
| writer.build(); |
| } catch (jthrowable ex) { |
| env->ExceptionDescribe(); |
| env->ExceptionClear(); |
| } |
| } |
| |
| |
| /** |
| * test write data |
| * test WithLoadOption interface |
| * |
| * @param env jni env |
| * @param path file path |
| * @param argc argument counter |
| * @param argv argument vector |
| * @return true or throw exception |
| */ |
| bool testWriteData(JNIEnv *env, char *path, int argc, char *argv[]) { |
| |
| char *jsonSchema = "[{stringField:string},{shortField:short},{intField:int},{longField:long},{doubleField:double},{boolField:boolean},{dateField:date},{timeField:timestamp},{floatField:float},{arrayField:array}]"; |
| try { |
| CarbonWriter writer; |
| writer.builder(env); |
| writer.outputPath(path); |
| writer.withCsvInput(jsonSchema); |
| writer.withLoadOption("complex_delimiter_level_1", "#"); |
| writer.withLoadOption("BAD_RECORDS_ACTION", "FORCE"); |
| writer.withLoadOption("bad_records_logger_enable", "FALSE"); |
| writer.writtenBy("CSDK"); |
| writer.taskNo(15541554.81); |
| writer.withThreadSafe(1); |
| writer.uniqueIdentifier(1549911814000000); |
| writer.withBlockSize(1); |
| writer.withBlockletSize(16); |
| writer.enableLocalDictionary(true); |
| writer.localDictionaryThreshold(10000); |
| if (argc > 3) { |
| writer.withHadoopConf("fs.s3a.access.key", argv[1]); |
| writer.withHadoopConf("fs.s3a.secret.key", argv[2]); |
| writer.withHadoopConf("fs.s3a.endpoint", argv[3]); |
| } |
| writer.build(); |
| |
| int rowNum = 70000; |
| int size = 10; |
| long longValue = 0; |
| double doubleValue = 0; |
| float floatValue = 0; |
| jclass objClass = env->FindClass("java/lang/String"); |
| for (int i = 0; i < rowNum; ++i) { |
| jobjectArray arr = env->NewObjectArray(size, objClass, 0); |
| char ctrInt[10]; |
| gcvt(i, 10, ctrInt); |
| |
| char a[15] = "robot"; |
| strcat(a, ctrInt); |
| jobject stringField = env->NewStringUTF(a); |
| env->SetObjectArrayElement(arr, 0, stringField); |
| |
| char ctrShort[10]; |
| gcvt(i % 10000, 10, ctrShort); |
| jobject shortField = env->NewStringUTF(ctrShort); |
| env->SetObjectArrayElement(arr, 1, shortField); |
| |
| jobject intField = env->NewStringUTF(ctrInt); |
| env->SetObjectArrayElement(arr, 2, intField); |
| |
| |
| char ctrLong[10]; |
| gcvt(longValue, 10, ctrLong); |
| longValue = longValue + 2; |
| jobject longField = env->NewStringUTF(ctrLong); |
| env->SetObjectArrayElement(arr, 3, longField); |
| |
| char ctrDouble[10]; |
| gcvt(doubleValue, 10, ctrDouble); |
| doubleValue = doubleValue + 2; |
| jobject doubleField = env->NewStringUTF(ctrDouble); |
| env->SetObjectArrayElement(arr, 4, doubleField); |
| |
| jobject boolField = env->NewStringUTF("true"); |
| env->SetObjectArrayElement(arr, 5, boolField); |
| |
| jobject dateField = env->NewStringUTF(" 2019-03-02"); |
| env->SetObjectArrayElement(arr, 6, dateField); |
| |
| jobject timeField = env->NewStringUTF("2019-02-12 03:03:34"); |
| env->SetObjectArrayElement(arr, 7, timeField); |
| |
| char ctrFloat[10]; |
| gcvt(floatValue, 10, ctrFloat); |
| floatValue = floatValue + 2; |
| jobject floatField = env->NewStringUTF(ctrFloat); |
| env->SetObjectArrayElement(arr, 8, floatField); |
| |
| jobject arrayField = env->NewStringUTF("Hello#World#From#Carbon"); |
| env->SetObjectArrayElement(arr, 9, arrayField); |
| |
| |
| writer.write(arr); |
| |
| env->DeleteLocalRef(stringField); |
| env->DeleteLocalRef(shortField); |
| env->DeleteLocalRef(intField); |
| env->DeleteLocalRef(longField); |
| env->DeleteLocalRef(doubleField); |
| env->DeleteLocalRef(floatField); |
| env->DeleteLocalRef(dateField); |
| env->DeleteLocalRef(timeField); |
| env->DeleteLocalRef(boolField); |
| env->DeleteLocalRef(arrayField); |
| env->DeleteLocalRef(arr); |
| } |
| writer.close(); |
| |
| CarbonReader carbonReader; |
| carbonReader.builder(env, path); |
| carbonReader.build(); |
| int i = 0; |
| int printNum = 10; |
| CarbonRow carbonRow(env); |
| while (carbonReader.hasNext()) { |
| jobject row = carbonReader.readNextRow(); |
| i++; |
| carbonRow.setCarbonRow(row); |
| if (i < printNum) { |
| printf("%s\t%d\t%ld\t", carbonRow.getString(0), carbonRow.getInt(1), carbonRow.getLong(2)); |
| jobjectArray array1 = carbonRow.getArray(3); |
| jsize length = env->GetArrayLength(array1); |
| int j = 0; |
| for (j = 0; j < length; j++) { |
| jobject element = env->GetObjectArrayElement(array1, j); |
| char *str = (char *) env->GetStringUTFChars((jstring) element, JNI_FALSE); |
| printf("%s\t", str); |
| } |
| printf("%d\t", carbonRow.getShort(4)); |
| printf("%d\t", carbonRow.getInt(5)); |
| printf("%ld\t", carbonRow.getLong(6)); |
| printf("%lf\t", carbonRow.getDouble(7)); |
| bool bool1 = carbonRow.getBoolean(8); |
| if (bool1) { |
| printf("true\t"); |
| } else { |
| printf("false\t"); |
| } |
| printf("%f\t\n", carbonRow.getFloat(9)); |
| } |
| env->DeleteLocalRef(row); |
| } |
| carbonReader.close(); |
| } catch (jthrowable ex) { |
| env->ExceptionDescribe(); |
| env->ExceptionClear(); |
| } |
| } |
| |
| void writeData(JNIEnv *env, CarbonWriter writer, int size, jclass objClass, char *stringField, short shortField) { |
| jobjectArray arr = env->NewObjectArray(size, objClass, 0); |
| |
| jobject jStringField = env->NewStringUTF(stringField); |
| env->SetObjectArrayElement(arr, 0, jStringField); |
| |
| char ctrShort[10]; |
| gcvt(shortField % 10000, 10, ctrShort); |
| jobject jShortField = env->NewStringUTF(ctrShort); |
| env->SetObjectArrayElement(arr, 1, jShortField); |
| |
| writer.write(arr); |
| |
| env->DeleteLocalRef(jStringField); |
| env->DeleteLocalRef(jShortField); |
| env->DeleteLocalRef(arr); |
| } |
| |
| /** |
| * test WithTableProperties interface |
| * |
| * @param env jni env |
| * @param path file path |
| * @param argc argument counter |
| * @param argv argument vector |
| * @return true or throw exception |
| */ |
| bool testWithTableProperty(JNIEnv *env, char *path, int argc, char **argv) { |
| |
| char *jsonSchema = "[{stringField:string},{shortField:short}]"; |
| try { |
| CarbonWriter writer; |
| writer.builder(env); |
| writer.outputPath(path); |
| writer.withCsvInput(jsonSchema); |
| writer.withTableProperty("sort_columns", "shortField"); |
| writer.enableLocalDictionary(false); |
| writer.writtenBy("CSDK"); |
| if (argc > 3) { |
| writer.withHadoopConf("fs.s3a.access.key", argv[1]); |
| writer.withHadoopConf("fs.s3a.secret.key", argv[2]); |
| writer.withHadoopConf("fs.s3a.endpoint", argv[3]); |
| } |
| writer.build(); |
| |
| int size = 10; |
| jclass objClass = env->FindClass("java/lang/String"); |
| |
| writeData(env, writer, size, objClass, "name3", 22); |
| writeData(env, writer, size, objClass, "name1", 11); |
| writeData(env, writer, size, objClass, "name2", 33); |
| writer.close(); |
| |
| CarbonReader carbonReader; |
| carbonReader.builder(env, path); |
| carbonReader.build(); |
| int i = 0; |
| CarbonRow carbonRow(env); |
| while (carbonReader.hasNext()) { |
| jobject row = carbonReader.readNextRow(); |
| i++; |
| carbonRow.setCarbonRow(row); |
| printf("%d\t%s\t\n", carbonRow.getShort(0), carbonRow.getString(1)); |
| env->DeleteLocalRef(row); |
| } |
| carbonReader.close(); |
| } catch (jthrowable ex) { |
| env->ExceptionDescribe(); |
| env->ExceptionClear(); |
| } |
| } |
| |
| /** |
| * test sortBy interface |
| * |
| * @param env jni env |
| * @param path file path |
| * @param argc argument counter |
| * @param argv argument vector |
| * @return true or throw exception |
| */ |
| bool testSortBy(JNIEnv *env, char *path, int argc, char **argv) { |
| |
| char *jsonSchema = "[{stringField:string},{shortField:short}]"; |
| char *sort[1]; |
| sort[0] = "shortField"; |
| try { |
| CarbonWriter writer; |
| writer.builder(env); |
| writer.outputPath(path); |
| writer.withCsvInput(jsonSchema); |
| writer.sortBy(1, sort); |
| writer.enableLocalDictionary(NULL); |
| writer.writtenBy("CSDK"); |
| if (argc > 3) { |
| writer.withHadoopConf("fs.s3a.access.key", argv[1]); |
| writer.withHadoopConf("fs.s3a.secret.key", argv[2]); |
| writer.withHadoopConf("fs.s3a.endpoint", argv[3]); |
| } |
| writer.build(); |
| |
| int size = 10; |
| jclass objClass = env->FindClass("java/lang/String"); |
| |
| writeData(env, writer, size, objClass, "name3", 22); |
| writeData(env, writer, size, objClass, "name1", 11); |
| writeData(env, writer, size, objClass, "name2", 33); |
| writer.close(); |
| |
| CarbonReader carbonReader; |
| carbonReader.builder(env, path); |
| carbonReader.build(); |
| int i = 0; |
| CarbonRow carbonRow(env); |
| while (carbonReader.hasNext()) { |
| jobject row = carbonReader.readNextRow(); |
| i++; |
| carbonRow.setCarbonRow(row); |
| printf("%d\t%s\t\n", carbonRow.getShort(0), carbonRow.getString(1)); |
| env->DeleteLocalRef(row); |
| } |
| carbonReader.close(); |
| } catch (jthrowable ex) { |
| env->ExceptionDescribe(); |
| env->ExceptionClear(); |
| } |
| } |
| |
| /** |
| * read data from S3 |
| * parameter is ak sk endpoint |
| * |
| * @param env jni env |
| * @param argv argument vector |
| * @return |
| */ |
| bool readFromS3(JNIEnv *env, char *path, char *argv[]) { |
| printf("\nRead data from S3:\n"); |
| CarbonReader reader; |
| |
| reader.builder(env, path, "test"); |
| reader.withHadoopConf("fs.s3a.access.key", argv[1]); |
| reader.withHadoopConf("fs.s3a.secret.key", argv[2]); |
| reader.withHadoopConf("fs.s3a.endpoint", argv[3]); |
| reader.build(); |
| printResult(env, reader); |
| } |
| |
| /** |
| * This a example for C++ interface to read carbon file |
| * If you want to test read data fromS3, please input the parameter: ak sk endpoint |
| * |
| * @param argc argument counter |
| * @param argv argument vector |
| * @return |
| */ |
| int main(int argc, char *argv[]) { |
| // init jvm |
| JNIEnv *env; |
| env = initJVM(); |
| char *S3WritePath = "s3a://csdk/WriterOutput/carbondata2"; |
| char *S3ReadPath = "s3a://csdk/WriterOutput/carbondata"; |
| |
| char *smallFilePath = "../../../../resources/carbondata"; |
| char *path = "../../../../../../../Downloads/carbon-data-big/dir2"; |
| char *S3Path = "s3a://csdk/bigData/i400bs128"; |
| |
| if (argc > 3) { |
| testWriteData(env, S3WritePath, 4, argv); |
| readSchema(env, S3WritePath, true, argv,4); |
| readSchema(env, S3WritePath, false, argv, 4); |
| readFromS3(env, S3ReadPath, argv); |
| testWithTableProperty(env, "s3a://csdk/dataProperty", 4, argv); |
| testSortBy(env, "s3a://csdk/dataSort", 4, argv); |
| |
| testReadNextRow(env, S3Path, 100000, argv, 4, false); |
| testReadNextRow(env, S3Path, 100000, argv, 4, true); |
| testReadNextBatchRow(env, S3Path, 100000, 100000, argv, 4, false); |
| testReadNextBatchRow(env, S3Path, 100000, 100000, argv, 4, true); |
| } else { |
| int batch = 32000; |
| int printNum = 32000; |
| |
| tryCatchException(env); |
| tryCarbonRowException(env, smallFilePath); |
| testCarbonProperties(env); |
| testValidateBadRecordsActionWithImproperValue(env, "./test"); |
| testValidateBadRecordsLoggerEnableWithImproperValue(env, "./test"); |
| testValidateQuoteCharWithImproperValue(env, "./test"); |
| testValidateEscapeCharWithImproperValue(env, "./test"); |
| testWriteData(env, "./data", 1, argv); |
| testWriteData(env, "./dataLoadOption", 1, argv); |
| readFromLocalWithoutProjection(env, smallFilePath); |
| readFromLocalWithProjection(env, smallFilePath); |
| testWithTableProperty(env, "./dataProperty", 1, argv); |
| testSortBy(env, "./dataSort", 1, argv); |
| readSchema(env, path, false, argv, 1); |
| readSchema(env, path, true, argv, 1); |
| |
| testReadNextRow(env, path, printNum, argv, 0, true); |
| testReadNextRow(env, path, printNum, argv, 0, false); |
| testReadNextBatchRow(env, path, batch, printNum, argv, 0, true); |
| testReadNextBatchRow(env, path, batch, printNum, argv, 0, false); |
| } |
| (jvm)->DestroyJavaVM(); |
| |
| cout << "\nfinish destroy jvm"; |
| return 0; |
| } |
| |