gaoran10 commented on a change in pull request #8422:
URL: https://github.com/apache/pulsar/pull/8422#discussion_r532660123
##########
File path:
pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java
##########
@@ -420,17 +435,92 @@ public boolean advanceNextPosition() {
//start time for deseralizing record
metricsTracker.start_RECORD_DESERIALIZE_TIME();
- if (this.schemaHandler instanceof KeyValueSchemaHandler) {
- ByteBuf keyByteBuf = null;
+ SchemaInfo schemaInfo;
+ try {
+ schemaInfo =
schemaInfoProvider.getSchemaByVersion(this.currentMessage.getSchemaVersion()).get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+
+ Map<ColumnHandle, FieldValueProvider> currentRowValuesMap = new
HashMap<>();
+
+ if (schemaInfo.getType().equals(SchemaType.KEY_VALUE)) {
+
+ PulsarRowDecoder keyDecoder =
decoderFactory.createRowDecoder(topicName,
Review comment:
It seems that the multi-version schema decoder cache could be added and
the decoder could be reused.
##########
File path:
pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/decoder/avro/PulsarAvroRowDecoderFactory.java
##########
@@ -0,0 +1,179 @@
+/**
+ * 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.pulsar.sql.presto.decoder.avro;
+
+import static com.google.common.collect.ImmutableList.toImmutableList;
+import static io.prestosql.spi.StandardErrorCode.NOT_SUPPORTED;
+import static io.prestosql.spi.type.DateType.DATE;
+import static io.prestosql.spi.type.TimeType.TIME;
+import static io.prestosql.spi.type.VarcharType.createUnboundedVarcharType;
+import static java.lang.String.format;
+import static java.util.stream.Collectors.toList;
+
+import com.google.common.collect.ImmutableList;
+import io.prestosql.decoder.DecoderColumnHandle;
+import io.prestosql.spi.PrestoException;
+import io.prestosql.spi.connector.ColumnMetadata;
+import io.prestosql.spi.type.ArrayType;
+import io.prestosql.spi.type.BigintType;
+import io.prestosql.spi.type.BooleanType;
+import io.prestosql.spi.type.DoubleType;
+import io.prestosql.spi.type.IntegerType;
+import io.prestosql.spi.type.RealType;
+import io.prestosql.spi.type.RowType;
+import io.prestosql.spi.type.StandardTypes;
+import io.prestosql.spi.type.TimestampType;
+import io.prestosql.spi.type.Type;
+import io.prestosql.spi.type.TypeManager;
+import io.prestosql.spi.type.TypeSignature;
+import io.prestosql.spi.type.TypeSignatureParameter;
+import io.prestosql.spi.type.VarbinaryType;
+import io.prestosql.spi.type.VarcharType;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+import org.apache.avro.LogicalType;
+import org.apache.avro.LogicalTypes;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaParseException;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericJsonSchema;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.schema.SchemaInfo;
+import org.apache.pulsar.sql.presto.PulsarColumnHandle;
+import org.apache.pulsar.sql.presto.PulsarColumnMetadata;
+import org.apache.pulsar.sql.presto.PulsarRowDecoder;
+import org.apache.pulsar.sql.presto.PulsarRowDecoderFactory;
+
+/**
+ * PulsarRowDecoderFactory for {@link
org.apache.pulsar.common.schema.SchemaType#AVRO}.
+ */
+public class PulsarAvroRowDecoderFactory implements PulsarRowDecoderFactory {
+
+ private TypeManager typeManager;
+
+ public PulsarAvroRowDecoderFactory(TypeManager typeManager) {
+ this.typeManager = typeManager;
+ }
+
+ @Override
+ public PulsarRowDecoder createRowDecoder(TopicName topicName, SchemaInfo
schemaInfo,
+ Set<DecoderColumnHandle> columns)
{
+ return new PulsarAvroRowDecoder((GenericAvroSchema)
GenericAvroSchema.of(schemaInfo), columns);
+ }
+
+ @Override
+ public List<ColumnMetadata> extractColumnMetadata(TopicName topicName,
SchemaInfo schemaInfo,
+
PulsarColumnHandle.HandleKeyValueType handleKeyValueType) {
+ String schemaJson = new String(schemaInfo.getSchema());
+ if (StringUtils.isBlank(schemaJson)) {
+ throw new PrestoException(NOT_SUPPORTED, "Topic "
+ + topicName.toString() + " does not have a valid schema");
+ }
+ Schema schema;
+ try {
+ schema = GenericJsonSchema.of(schemaInfo).getAvroSchema();
+ } catch (SchemaParseException ex) {
+ throw new PrestoException(NOT_SUPPORTED, "Topic "
+ + topicName.toString() + " does not have a valid schema");
+ }
+
+ //TODO : check schema cyclic definitions which may case
java.lang.StackOverflowError
Review comment:
Does this problem is fixed?
##########
File path:
pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java
##########
@@ -420,17 +435,92 @@ public boolean advanceNextPosition() {
//start time for deseralizing record
metricsTracker.start_RECORD_DESERIALIZE_TIME();
- if (this.schemaHandler instanceof KeyValueSchemaHandler) {
- ByteBuf keyByteBuf = null;
+ SchemaInfo schemaInfo;
+ try {
+ schemaInfo =
schemaInfoProvider.getSchemaByVersion(this.currentMessage.getSchemaVersion()).get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+
+ Map<ColumnHandle, FieldValueProvider> currentRowValuesMap = new
HashMap<>();
+
+ if (schemaInfo.getType().equals(SchemaType.KEY_VALUE)) {
+
+ PulsarRowDecoder keyDecoder =
decoderFactory.createRowDecoder(topicName,
+ schemaInfo,
+ columnHandles.stream()
+ .filter(col -> !col.isInternal())
+ .filter(col ->
PulsarColumnHandle.HandleKeyValueType.KEY
+ .equals(col.getHandleKeyValueType()))
+ .collect(toImmutableSet()));
+
+ PulsarRowDecoder messageDecoder =
decoderFactory.createRowDecoder(topicName,
Review comment:
same as above.
##########
File path:
pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarConnector.java
##########
@@ -110,16 +81,20 @@
protected Map<TopicName, PulsarRecordCursor> pulsarRecordCursors = new
HashMap<>();
+ protected static PulsarDispatchingRowDecoderFactory
dispatchingRowDecoderFactory;
+
protected final static PulsarConnectorId pulsarConnectorId = new
PulsarConnectorId("test-connector");
- protected static List<TopicName> topicNames;
- protected static List<TopicName> partitionedTopicNames;
+ protected static List<TopicName> topicNames;
+ protected static List<TopicName> partitionedTopicNames;
Review comment:
```suggestion
protected static List<TopicName> partitionedTopicNames;
```
##########
File path:
pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/decoder/avro/PulsarAvroRowDecoderFactory.java
##########
@@ -0,0 +1,179 @@
+/**
+ * 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.pulsar.sql.presto.decoder.avro;
+
+import static com.google.common.collect.ImmutableList.toImmutableList;
+import static io.prestosql.spi.StandardErrorCode.NOT_SUPPORTED;
+import static io.prestosql.spi.type.DateType.DATE;
+import static io.prestosql.spi.type.TimeType.TIME;
+import static io.prestosql.spi.type.VarcharType.createUnboundedVarcharType;
+import static java.lang.String.format;
+import static java.util.stream.Collectors.toList;
+
+import com.google.common.collect.ImmutableList;
+import io.prestosql.decoder.DecoderColumnHandle;
+import io.prestosql.spi.PrestoException;
+import io.prestosql.spi.connector.ColumnMetadata;
+import io.prestosql.spi.type.ArrayType;
+import io.prestosql.spi.type.BigintType;
+import io.prestosql.spi.type.BooleanType;
+import io.prestosql.spi.type.DoubleType;
+import io.prestosql.spi.type.IntegerType;
+import io.prestosql.spi.type.RealType;
+import io.prestosql.spi.type.RowType;
+import io.prestosql.spi.type.StandardTypes;
+import io.prestosql.spi.type.TimestampType;
+import io.prestosql.spi.type.Type;
+import io.prestosql.spi.type.TypeManager;
+import io.prestosql.spi.type.TypeSignature;
+import io.prestosql.spi.type.TypeSignatureParameter;
+import io.prestosql.spi.type.VarbinaryType;
+import io.prestosql.spi.type.VarcharType;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+import org.apache.avro.LogicalType;
+import org.apache.avro.LogicalTypes;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaParseException;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericJsonSchema;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.schema.SchemaInfo;
+import org.apache.pulsar.sql.presto.PulsarColumnHandle;
+import org.apache.pulsar.sql.presto.PulsarColumnMetadata;
+import org.apache.pulsar.sql.presto.PulsarRowDecoder;
+import org.apache.pulsar.sql.presto.PulsarRowDecoderFactory;
+
+/**
+ * PulsarRowDecoderFactory for {@link
org.apache.pulsar.common.schema.SchemaType#AVRO}.
+ */
+public class PulsarAvroRowDecoderFactory implements PulsarRowDecoderFactory {
+
+ private TypeManager typeManager;
+
+ public PulsarAvroRowDecoderFactory(TypeManager typeManager) {
+ this.typeManager = typeManager;
+ }
+
+ @Override
+ public PulsarRowDecoder createRowDecoder(TopicName topicName, SchemaInfo
schemaInfo,
+ Set<DecoderColumnHandle> columns)
{
+ return new PulsarAvroRowDecoder((GenericAvroSchema)
GenericAvroSchema.of(schemaInfo), columns);
+ }
+
+ @Override
+ public List<ColumnMetadata> extractColumnMetadata(TopicName topicName,
SchemaInfo schemaInfo,
+
PulsarColumnHandle.HandleKeyValueType handleKeyValueType) {
+ String schemaJson = new String(schemaInfo.getSchema());
+ if (StringUtils.isBlank(schemaJson)) {
+ throw new PrestoException(NOT_SUPPORTED, "Topic "
+ + topicName.toString() + " does not have a valid schema");
+ }
+ Schema schema;
+ try {
+ schema = GenericJsonSchema.of(schemaInfo).getAvroSchema();
+ } catch (SchemaParseException ex) {
+ throw new PrestoException(NOT_SUPPORTED, "Topic "
+ + topicName.toString() + " does not have a valid schema");
+ }
+
+ //TODO : check schema cyclic definitions which may case
java.lang.StackOverflowError
+
+ return schema.getFields().stream()
+ .map(field ->
+ new PulsarColumnMetadata(field.name(),
parseAvroPrestoType(
+ field.name(), field.schema()),
field.schema().toString(), null, false, false,
+ handleKeyValueType, new
PulsarColumnMetadata.DecoderExtraInfo(field.name(),
+ null, null))
+
+ ).collect(toList());
+ }
+
+ private Type parseAvroPrestoType(String fieldname, Schema schema) {
+ Schema.Type type = schema.getType();
+ LogicalType logicalType = schema.getLogicalType();
+ switch (type) {
+ case STRING:
+ case ENUM:
+ return createUnboundedVarcharType();
+ case NULL:
+ throw new UnsupportedOperationException(
+ format("field '%s' NULL type code should not be
reached,"
+ + "please check the schema or report the
bug.", fieldname));
+ case FIXED:
+ case BYTES:
+ //TODO: support decimal logicalType
+ return VarbinaryType.VARBINARY;
+ case INT:
+ if (logicalType == LogicalTypes.timeMillis()) {
+ return TIME;
+ } else if (logicalType == LogicalTypes.date()) {
+ return DATE;
+ }
+ return IntegerType.INTEGER;
+ case LONG:
+ if (logicalType == LogicalTypes.timestampMillis()) {
+ return TimestampType.TIMESTAMP;
+ }
+ //TODO: support timestamp_microseconds logicalType :
https://github.com/prestosql/presto/issues/1284
Review comment:
same as above.
##########
File path:
pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarConnector.java
##########
@@ -179,28 +150,13 @@
public static class Bar {
public Integer field1;
public String field2;
- public Boo test;
Review comment:
Is there any test to test the nest object query?
##########
File path:
pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java
##########
@@ -420,17 +435,92 @@ public boolean advanceNextPosition() {
//start time for deseralizing record
metricsTracker.start_RECORD_DESERIALIZE_TIME();
- if (this.schemaHandler instanceof KeyValueSchemaHandler) {
- ByteBuf keyByteBuf = null;
+ SchemaInfo schemaInfo;
+ try {
+ schemaInfo =
schemaInfoProvider.getSchemaByVersion(this.currentMessage.getSchemaVersion()).get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+
+ Map<ColumnHandle, FieldValueProvider> currentRowValuesMap = new
HashMap<>();
+
+ if (schemaInfo.getType().equals(SchemaType.KEY_VALUE)) {
+
+ PulsarRowDecoder keyDecoder =
decoderFactory.createRowDecoder(topicName,
+ schemaInfo,
+ columnHandles.stream()
+ .filter(col -> !col.isInternal())
+ .filter(col ->
PulsarColumnHandle.HandleKeyValueType.KEY
+ .equals(col.getHandleKeyValueType()))
+ .collect(toImmutableSet()));
+
+ PulsarRowDecoder messageDecoder =
decoderFactory.createRowDecoder(topicName,
+ schemaInfo,
+ columnHandles.stream()
+ .filter(col -> !col.isInternal())
+ .filter(col ->
PulsarColumnHandle.HandleKeyValueType.VALUE
+ .equals(col.getHandleKeyValueType()))
+ .collect(toImmutableSet()));
+
+ Optional<Map<DecoderColumnHandle, FieldValueProvider>> decodedKey;
if (this.currentMessage.getKeyBytes().isPresent()) {
- keyByteBuf = this.currentMessage.getKeyBytes().get();
+ decodedKey =
keyDecoder.decodeRow(this.currentMessage.getKeyBytes().get());
+ decodedKey.ifPresent(currentRowValuesMap::putAll);
}
- currentRecord = this.schemaHandler.deserialize(keyByteBuf,
- this.currentMessage.getData(),
this.currentMessage.getSchemaVersion());
+ Optional<Map<DecoderColumnHandle, FieldValueProvider>>
decodedValue =
+ messageDecoder.decodeRow(this.currentMessage.getData());
+ decodedValue.ifPresent(currentRowValuesMap::putAll);
} else {
- currentRecord =
this.schemaHandler.deserialize(this.currentMessage.getData(),
- this.currentMessage.getSchemaVersion());
+ PulsarRowDecoder messageDecoder =
decoderFactory.createRowDecoder(topicName,
+ schemaInfo,
+ columnHandles.stream()
+ .filter(col -> !col.isInternal())
+ .filter(col ->
PulsarColumnHandle.HandleKeyValueType.NONE
+ .equals(col.getHandleKeyValueType()))
+ .collect(toImmutableSet()));
+ Optional<Map<DecoderColumnHandle, FieldValueProvider>>
decodedValue =
+ messageDecoder.decodeRow(this.currentMessage.getData());
+ decodedValue.ifPresent(currentRowValuesMap::putAll);
+ }
+
+ for (DecoderColumnHandle columnHandle : columnHandles) {
+ if (columnHandle.isInternal()) {
+ if
(PulsarInternalColumn.PARTITION.getName().equals(columnHandle.getName())) {
Review comment:
The switch is more efficient and the `PulsarInternalColumn` could be
changed to an enum.
##########
File path:
pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/decoder/json/PulsarJsonRowDecoderFactory.java
##########
@@ -0,0 +1,174 @@
+/**
+ * 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.pulsar.sql.presto.decoder.json;
+
+import static com.google.common.collect.ImmutableList.toImmutableList;
+import static io.prestosql.spi.StandardErrorCode.NOT_SUPPORTED;
+import static io.prestosql.spi.type.DateType.DATE;
+import static io.prestosql.spi.type.TimeType.TIME;
+import static io.prestosql.spi.type.VarcharType.createUnboundedVarcharType;
+import static java.lang.String.format;
+import static java.util.stream.Collectors.toList;
+
+import com.google.common.collect.ImmutableList;
+import io.prestosql.decoder.DecoderColumnHandle;
+import io.prestosql.spi.PrestoException;
+import io.prestosql.spi.connector.ColumnMetadata;
+import io.prestosql.spi.type.ArrayType;
+import io.prestosql.spi.type.BigintType;
+import io.prestosql.spi.type.BooleanType;
+import io.prestosql.spi.type.DoubleType;
+import io.prestosql.spi.type.IntegerType;
+import io.prestosql.spi.type.RealType;
+import io.prestosql.spi.type.RowType;
+import io.prestosql.spi.type.StandardTypes;
+import io.prestosql.spi.type.TimestampType;
+import io.prestosql.spi.type.Type;
+import io.prestosql.spi.type.TypeManager;
+import io.prestosql.spi.type.TypeSignature;
+import io.prestosql.spi.type.TypeSignatureParameter;
+import io.prestosql.spi.type.VarbinaryType;
+import io.prestosql.spi.type.VarcharType;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import org.apache.avro.LogicalType;
+import org.apache.avro.LogicalTypes;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaParseException;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.client.impl.schema.generic.GenericJsonSchema;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.schema.SchemaInfo;
+import org.apache.pulsar.sql.presto.PulsarColumnHandle;
+import org.apache.pulsar.sql.presto.PulsarColumnMetadata;
+import org.apache.pulsar.sql.presto.PulsarRowDecoderFactory;
+
+/**
+ * PulsarRowDecoderFactory for {@link
org.apache.pulsar.common.schema.SchemaType#JSON}.
+ */
+public class PulsarJsonRowDecoderFactory implements PulsarRowDecoderFactory {
+
+ private TypeManager typeManager;
+
+ public PulsarJsonRowDecoderFactory(TypeManager typeManager) {
+ this.typeManager = typeManager;
+ }
+
+ @Override
+ public PulsarJsonRowDecoder createRowDecoder(TopicName topicName,
SchemaInfo schemaInfo,
+ Set<DecoderColumnHandle>
columns) {
+ return new PulsarJsonRowDecoder((GenericJsonSchema)
GenericJsonSchema.of(schemaInfo), columns);
+ }
+
+ @Override
+ public List<ColumnMetadata> extractColumnMetadata(TopicName topicName,
SchemaInfo schemaInfo,
+
PulsarColumnHandle.HandleKeyValueType handleKeyValueType) {
+ String schemaJson = new String(schemaInfo.getSchema());
+ if (StringUtils.isBlank(schemaJson)) {
+ throw new PrestoException(NOT_SUPPORTED, "Topic "
+ + topicName.toString() + " does not have a valid schema");
+ }
+ Schema schema;
+ try {
+ schema = GenericJsonSchema.of(schemaInfo).getAvroSchema();
+ } catch (SchemaParseException ex) {
+ throw new PrestoException(NOT_SUPPORTED, "Topic "
+ + topicName.toString() + " does not have a valid schema");
+ }
+
+ //TODO : check schema cyclic definitions which may case
java.lang.StackOverflowError
Review comment:
Does this problem is fixed?
##########
File path:
pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarConnector.java
##########
@@ -110,16 +81,20 @@
protected Map<TopicName, PulsarRecordCursor> pulsarRecordCursors = new
HashMap<>();
+ protected static PulsarDispatchingRowDecoderFactory
dispatchingRowDecoderFactory;
+
protected final static PulsarConnectorId pulsarConnectorId = new
PulsarConnectorId("test-connector");
- protected static List<TopicName> topicNames;
- protected static List<TopicName> partitionedTopicNames;
+ protected static List<TopicName> topicNames;
Review comment:
```suggestion
protected static List<TopicName> topicNames;
```
##########
File path:
pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/decoder/avro/PulsarAvroRowDecoderFactory.java
##########
@@ -0,0 +1,179 @@
+/**
+ * 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.pulsar.sql.presto.decoder.avro;
+
+import static com.google.common.collect.ImmutableList.toImmutableList;
+import static io.prestosql.spi.StandardErrorCode.NOT_SUPPORTED;
+import static io.prestosql.spi.type.DateType.DATE;
+import static io.prestosql.spi.type.TimeType.TIME;
+import static io.prestosql.spi.type.VarcharType.createUnboundedVarcharType;
+import static java.lang.String.format;
+import static java.util.stream.Collectors.toList;
+
+import com.google.common.collect.ImmutableList;
+import io.prestosql.decoder.DecoderColumnHandle;
+import io.prestosql.spi.PrestoException;
+import io.prestosql.spi.connector.ColumnMetadata;
+import io.prestosql.spi.type.ArrayType;
+import io.prestosql.spi.type.BigintType;
+import io.prestosql.spi.type.BooleanType;
+import io.prestosql.spi.type.DoubleType;
+import io.prestosql.spi.type.IntegerType;
+import io.prestosql.spi.type.RealType;
+import io.prestosql.spi.type.RowType;
+import io.prestosql.spi.type.StandardTypes;
+import io.prestosql.spi.type.TimestampType;
+import io.prestosql.spi.type.Type;
+import io.prestosql.spi.type.TypeManager;
+import io.prestosql.spi.type.TypeSignature;
+import io.prestosql.spi.type.TypeSignatureParameter;
+import io.prestosql.spi.type.VarbinaryType;
+import io.prestosql.spi.type.VarcharType;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+import org.apache.avro.LogicalType;
+import org.apache.avro.LogicalTypes;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaParseException;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericJsonSchema;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.schema.SchemaInfo;
+import org.apache.pulsar.sql.presto.PulsarColumnHandle;
+import org.apache.pulsar.sql.presto.PulsarColumnMetadata;
+import org.apache.pulsar.sql.presto.PulsarRowDecoder;
+import org.apache.pulsar.sql.presto.PulsarRowDecoderFactory;
+
+/**
+ * PulsarRowDecoderFactory for {@link
org.apache.pulsar.common.schema.SchemaType#AVRO}.
+ */
+public class PulsarAvroRowDecoderFactory implements PulsarRowDecoderFactory {
+
+ private TypeManager typeManager;
+
+ public PulsarAvroRowDecoderFactory(TypeManager typeManager) {
+ this.typeManager = typeManager;
+ }
+
+ @Override
+ public PulsarRowDecoder createRowDecoder(TopicName topicName, SchemaInfo
schemaInfo,
+ Set<DecoderColumnHandle> columns)
{
+ return new PulsarAvroRowDecoder((GenericAvroSchema)
GenericAvroSchema.of(schemaInfo), columns);
+ }
+
+ @Override
+ public List<ColumnMetadata> extractColumnMetadata(TopicName topicName,
SchemaInfo schemaInfo,
+
PulsarColumnHandle.HandleKeyValueType handleKeyValueType) {
+ String schemaJson = new String(schemaInfo.getSchema());
+ if (StringUtils.isBlank(schemaJson)) {
+ throw new PrestoException(NOT_SUPPORTED, "Topic "
+ + topicName.toString() + " does not have a valid schema");
+ }
+ Schema schema;
+ try {
+ schema = GenericJsonSchema.of(schemaInfo).getAvroSchema();
+ } catch (SchemaParseException ex) {
+ throw new PrestoException(NOT_SUPPORTED, "Topic "
+ + topicName.toString() + " does not have a valid schema");
+ }
+
+ //TODO : check schema cyclic definitions which may case
java.lang.StackOverflowError
+
+ return schema.getFields().stream()
+ .map(field ->
+ new PulsarColumnMetadata(field.name(),
parseAvroPrestoType(
+ field.name(), field.schema()),
field.schema().toString(), null, false, false,
+ handleKeyValueType, new
PulsarColumnMetadata.DecoderExtraInfo(field.name(),
+ null, null))
+
+ ).collect(toList());
+ }
+
+ private Type parseAvroPrestoType(String fieldname, Schema schema) {
+ Schema.Type type = schema.getType();
+ LogicalType logicalType = schema.getLogicalType();
+ switch (type) {
+ case STRING:
+ case ENUM:
+ return createUnboundedVarcharType();
+ case NULL:
+ throw new UnsupportedOperationException(
+ format("field '%s' NULL type code should not be
reached,"
+ + "please check the schema or report the
bug.", fieldname));
+ case FIXED:
+ case BYTES:
+ //TODO: support decimal logicalType
Review comment:
This feature will be added in the future, right?
##########
File path:
pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarConnector.java
##########
@@ -110,16 +81,20 @@
protected Map<TopicName, PulsarRecordCursor> pulsarRecordCursors = new
HashMap<>();
+ protected static PulsarDispatchingRowDecoderFactory
dispatchingRowDecoderFactory;
+
protected final static PulsarConnectorId pulsarConnectorId = new
PulsarConnectorId("test-connector");
- protected static List<TopicName> topicNames;
- protected static List<TopicName> partitionedTopicNames;
+ protected static List<TopicName> topicNames;
+ protected static List<TopicName> partitionedTopicNames;
protected static Map<String, Integer> partitionedTopicsToPartitions;
protected static Map<String, SchemaInfo> topicsToSchemas;
protected static Map<String, Long> topicsToNumEntries;
private final static ObjectMapper objectMapper = new ObjectMapper();
+ protected static List<String> fooFieldNames = new ArrayList<>();
Review comment:
```suggestion
protected static List<String> fooFieldNames = new ArrayList<>();
```
##########
File path:
pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarKeyValueSchemaHandler.java
##########
@@ -1,353 +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.pulsar.sql.presto;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
-import io.prestosql.spi.connector.ColumnMetadata;
-import java.io.IOException;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Objects;
-import java.util.Optional;
-import lombok.Data;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.client.api.Schema;
-import org.apache.pulsar.client.impl.schema.KeyValueSchema;
-import org.apache.pulsar.client.impl.schema.KeyValueSchemaInfo;
-import org.apache.pulsar.common.api.raw.RawMessage;
-import org.apache.pulsar.common.api.raw.RawMessageImpl;
-import org.apache.pulsar.common.naming.TopicName;
-import org.apache.pulsar.common.schema.KeyValue;
-import org.apache.pulsar.common.schema.KeyValueEncodingType;
-import org.apache.pulsar.common.schema.SchemaInfo;
-import org.mockito.Mockito;
-import org.testng.Assert;
-import org.testng.annotations.Test;
-
-import static org.mockito.Mockito.mock;
-
-
-/**
- * Unit test for KeyValueSchemaHandler
- */
-@Slf4j
-public class TestPulsarKeyValueSchemaHandler {
Review comment:
Could you add a test for `KeyValueSchema` datas?
##########
File path:
pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarPrimitiveSchemaHandler.java
##########
@@ -1,164 +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.pulsar.sql.presto;
-
-import io.netty.buffer.ByteBufAllocator;
-import io.prestosql.spi.connector.ColumnMetadata;
-import lombok.extern.slf4j.Slf4j;
-
-import org.apache.pulsar.client.impl.schema.BooleanSchema;
-import org.apache.pulsar.client.impl.schema.ByteSchema;
-import org.apache.pulsar.client.impl.schema.BytesSchema;
-import org.apache.pulsar.client.impl.schema.DateSchema;
-import org.apache.pulsar.client.impl.schema.DoubleSchema;
-import org.apache.pulsar.client.impl.schema.FloatSchema;
-import org.apache.pulsar.client.impl.schema.IntSchema;
-import org.apache.pulsar.client.impl.schema.LongSchema;
-import org.apache.pulsar.client.impl.schema.ShortSchema;
-import org.apache.pulsar.client.impl.schema.StringSchema;
-import org.apache.pulsar.client.impl.schema.TimeSchema;
-import org.apache.pulsar.client.impl.schema.TimestampSchema;
-import org.apache.pulsar.common.api.raw.RawMessage;
-import org.apache.pulsar.common.naming.TopicName;
-import org.apache.pulsar.common.schema.SchemaInfo;
-import org.apache.pulsar.common.schema.SchemaType;
-import org.testng.Assert;
-import org.testng.annotations.Test;
-
-import java.sql.Time;
-import java.sql.Timestamp;
-import java.util.Date;
-import java.util.List;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-@Slf4j
-public class TestPulsarPrimitiveSchemaHandler {
Review comment:
Could you add a test for `PulsarPrimitiveRowDecoder`?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
