amrishlal commented on a change in pull request #8426:
URL: https://github.com/apache/pinot/pull/8426#discussion_r837995097
##########
File path:
pinot-plugins/pinot-input-format/pinot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroSchemaUtil.java
##########
@@ -95,6 +100,9 @@ public static ObjectNode toAvroSchemaJsonObject(FieldSpec
fieldSpec) {
case BYTES:
jsonSchema.set("type", convertStringsToJsonArray("null", "bytes"));
return jsonSchema;
+ case JSON:
+ jsonSchema.set("type", convertStringsToJsonArray("null", "json"));
Review comment:
Done
##########
File path:
pinot-plugins/pinot-input-format/pinot-avro-base/src/main/java/org/apache/pinot/plugin/inputformat/avro/AvroSchemaUtil.java
##########
@@ -50,6 +50,11 @@ public static DataType valueOf(Schema.Type avroType) {
return DataType.STRING;
case BYTES:
Review comment:
Done
##########
File path:
pinot-core/src/test/java/org/apache/pinot/queries/JsonIngestionFromAvroQueriesTest.java
##########
@@ -0,0 +1,341 @@
+/**
+ * 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.pinot.queries;
+
+import com.google.common.collect.Lists;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.avro.Schema;
+import org.apache.avro.file.DataFileWriter;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.core.common.Operator;
+import org.apache.pinot.core.operator.blocks.IntermediateResultsBlock;
+import org.apache.pinot.plugin.inputformat.avro.AvroRecordReader;
+import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
+import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.spi.ImmutableSegment;
+import org.apache.pinot.segment.spi.IndexSegment;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.data.readers.RecordReader;
+import org.apache.pinot.spi.utils.Pair;
+import org.apache.pinot.spi.utils.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.apache.avro.Schema.*;
+
+
+/**
+ * Test if ComplexType (RECORD, ARRAY, MAP, UNION, ENUM, and FIXED) field from
an AVRO file can be ingested into a JSON
+ * column in a Pinot segment.
+ */
+public class JsonIngestionFromAvroQueriesTest extends BaseQueriesTest {
+ private static final File INDEX_DIR = new File(FileUtils.getTempDirectory(),
"JsonIngestionFromAvroTest");
+ private static final File AVRO_DATA_FILE = new File(INDEX_DIR,
"JsonIngestionFromAvroTest.avro");
+ private static final String RAW_TABLE_NAME = "testTable";
+ private static final String SEGMENT_NAME = "testSegment";
+
+ private static final String INT_COLUMN = "intColumn";
+ private static final String JSON_COLUMN_1 = "jsonColumn1"; // for testing
RECORD, ARRAY, MAP, UNION
+ private static final String JSON_COLUMN_2 = "jsonColumn2"; // for testing
ENUM
+ private static final String JSON_COLUMN_3 = "jsonColumn3"; // for testing
FIXED
+ private static final String STRING_COLUMN = "stringColumn";
+ private static final org.apache.pinot.spi.data.Schema SCHEMA =
+ new
org.apache.pinot.spi.data.Schema.SchemaBuilder().addSingleValueDimension(INT_COLUMN,
FieldSpec.DataType.INT)
+ .addSingleValueDimension(JSON_COLUMN_1, FieldSpec.DataType.JSON)
+ .addSingleValueDimension(JSON_COLUMN_2, FieldSpec.DataType.JSON)
+ .addSingleValueDimension(JSON_COLUMN_3, FieldSpec.DataType.JSON)
+ .addSingleValueDimension(STRING_COLUMN,
FieldSpec.DataType.STRING).build();
+ private static final TableConfig TABLE_CONFIG =
+ new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).build();
+
+ private IndexSegment _indexSegment;
+ private List<IndexSegment> _indexSegments;
+
+ @Override
+ protected String getFilter() {
+ return "";
+ }
+
+ @Override
+ protected IndexSegment getIndexSegment() {
+ return _indexSegment;
+ }
+
+ @Override
+ protected List<IndexSegment> getIndexSegments() {
+ return _indexSegments;
+ }
+
+ /** @return {@link GenericRow} representing a row in Pinot table. */
+ private static GenericRow createTableRecord(int intValue, String
stringValue, Object jsonValue,
+ GenericData.EnumSymbol enumValue, GenericData.Fixed fixedValue) {
+ GenericRow record = new GenericRow();
+ record.putValue(INT_COLUMN, intValue);
+ record.putValue(STRING_COLUMN, stringValue);
+ record.putValue(JSON_COLUMN_1, jsonValue);
+ record.putValue(JSON_COLUMN_2, enumValue);
+ record.putValue(JSON_COLUMN_3, fixedValue);
+ return record;
+ }
+
+ private static Map<String, String> createMapField(Pair<String, String>[]
pairs) {
+ Map<String, String> map = new HashMap<>();
+ for (Pair<String, String> pair : pairs) {
+ map.put(pair.getFirst(), pair.getSecond());
+ }
+ return map;
+ }
+
+ private static Schema createRecordSchema() {
+ List<Field> fields = new ArrayList<>();
+ fields.add(new Field("id", create(Type.INT)));
+ fields.add(new Field("name", create(Type.STRING)));
+ return createRecord("record", "doc",
JsonIngestionFromAvroQueriesTest.class.getCanonicalName(), false, fields);
+ }
+
+ private static GenericData.Record createRecordField(String k1, int v1,
String k2, String v2) {
+ GenericData.Record record = new GenericData.Record(createRecordSchema());
+ record.put(k1, v1);
+ record.put(k2, v2);
+ return record;
+ }
+
+ private static GenericData.EnumSymbol createEnumField(Schema enumSchema,
String enumValue) {
+ return new GenericData.EnumSymbol(enumSchema, enumValue);
+ }
+
+ private static GenericData.Fixed createFixedField(Schema fixedSchema, int
value) {
+ byte[] bytes = {(byte) (value >> 24), (byte) (value >> 16), (byte) (value
>> 8), (byte) value};
+ return new GenericData.Fixed(fixedSchema, bytes);
+ }
+
+ private static void createInputFile()
+ throws IOException {
+ INDEX_DIR.mkdir();
+ Schema avroSchema = Schema.createRecord("eventsRecord", null, null, false);
Review comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
