Hi all, I'm pushing a table from postgres to kafka (using debezium) to Pinot. The table has a few geography columns. When creating the realtime table however, I get am getting error on Pinot (below).

java.lang.IllegalStateException: Cannot read single-value from Collection: [AQEAACDmEAAA5no2BviTXcB1T2ijhAxBQA==, 4326] for column: point
	at shaded.com.google.common.base.Preconditions.checkState(Preconditions.java:721) ~[pinot-all-0.7.1-jar-with-dependencies.jar:0.7.1-afa4b252ab1c424ddd6c859bb305b2aa342b66ed]
	at org.apache.pinot.core.data.recordtransformer.DataTypeTransformer.standardizeCollection(DataTypeTransformer.java:193) ~[pinot-all-0.7.1-jar-with-dependencies.jar:0.7.1-afa4b252ab1c424ddd6c859bb305b2aa342b66ed]
	at org.apache.pinot.core.data.recordtransformer.DataTypeTransformer.standardize(DataTypeTransformer.java:138) ~[pinot-all-0.7.1-jar-with-dependencies.jar:0.7.1-afa4b252ab1c424ddd6c859bb305b2aa342b66ed]
	at org.apache.pinot.core.data.recordtransformer.DataTypeTransformer.transform(DataTypeTransformer.java:88) ~[pinot-all-0.7.1-jar-with-dependencies.jar:0.7.1-afa4b252ab1c424ddd6c859bb305b2aa342b66ed]
	at org.apache.pinot.core.data.recordtransformer.CompositeTransformer.transform(CompositeTransformer.java:82) ~[pinot-all-0.7.1-jar-with-dependencies.jar:0.7.1-afa4b252ab1c424ddd6c859bb305b2aa342b66ed]
	at org.apache.pinot.core.data.manager.realtime.LLRealtimeSegmentDataManager.processStreamEvents(LLRealtimeSegmentDataManager.java:491) [pinot-all-0.7.1-jar-with-dependencies.jar:0.7.1-afa4b252ab1c424ddd6c859bb305b2aa342b66ed]
	at org.apache.pinot.core.data.manager.realtime.LLRealtimeSegmentDataManager.consumeLoop(LLRealtimeSegmentDataManager.java:402) [pinot-all-0.7.1-jar-with-dependencies.jar:0.7.1-afa4b252ab1c424ddd6c859bb305b2aa342b66ed]
	at org.apache.pinot.core.data.manager.realtime.LLRealtimeSegmentDataManager$PartitionConsumer.run(LLRealtimeSegmentDataManager.java:538) [pinot-all-0.7.1-jar-with-dependencies.jar:0.7.1-afa4b252ab1c424ddd6c859bb305b2aa342b66ed]
	at java.lang.Thread.run(Thread.java:748) [?:1.8.0_282]

The point column has this as value:

"point" : {
      "wkb" : "AQEAACDmEAAA5no2BviTXcB1T2ijhAxBQA==",
      "srid" : 4326
    },

Any suggestions on how to resolve? I have the column as string in the Pinot table schema.

Can you please share the schema? Are you planning to add geo index to this column?

try

ST_GeogFromWKB

from https://docs.pinot.apache.org/basics/indexing/geospatial-support#constructors

You should be able to decode the input string using the in-built

base64Decode()

function. See https://docs.pinot.apache.org/developers/advanced/ingestion-level-transformations on how to use ingestion transforms

Thank you for the reply. I do plan on adding geo index to the column. Below is the schema. I used this example.

{
  "schemaName": "table",
  "primaryKeyColumns": [
    "id"
  ],
  "dimensionFieldSpecs": [
    {
      "name": "id",
      "dataType": "LONG"
    },
    {
      "name": "title",
      "dataType": "STRING"
    },
    {
      "name": "state",
      "dataType": "STRING"
    },
    {
      "name": "latitude",
      "dataType": "DOUBLE",
      "defaultValue": 0
    },
    {
      "name": "longitude",
      "dataType": "DOUBLE",
      "defaultValue": 0
    },
    {
      "name": "point",
      "dataType": "STRING",
      "defaultValue": ""
    },
    {
      "dataType": "BYTES",
      "name": "location_st_point",
      "transformFunction": "toSphericalGeography(<not sure what goes here>(point))"
    }
  ],
  "metricFieldSpecs": [
    {
      "name": "amount",
      "dataType": "DOUBLE",
      "defaultNullValue": 0
    }
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "created_at",
      "dataType": "LONG",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:MILLISECONDS"
    }
  ]
}

@Kamal Chavda Yes. Does it work?

Hi @Jackie, it doesn't work. I am getting this error:

Getting Helix leader: 172.22.0.5_9000, Helix version: 0.9.8, mtime: 1633982847545
pinot_1      | Caught exception while compiling SQL query: select toGeometry(base64Decode('AQEAACDmEAAAT0wojs27YsA0a4TZX5hOQA=='))from meetupRsvp limit 10
pinot_1      | org.apache.pinot.sql.parsers.SqlCompilationException: Caught exception while invoking method: public static byte[] org.apache.pinot.core.geospatial.transform.function.ScalarFunctions.toGeometry(byte[]) with arguments: [[B@1e36f6bb]
pinot_1      | 	at org.apache.pinot.sql.parsers.CalciteSqlParser.invokeCompileTimeFunctionExpression(CalciteSqlParser.java:1116) ~[pinot-all-0.8.0-jar-with-dependencies.jar:0.8.0-c4ceff06d21fc1c1b88469a8dbae742a4b609808]
pinot_1      | 	at org.apache.pinot.sql.parsers.CalciteSqlParser.invokeCompileTimeFunctions(CalciteSqlParser.java:578) ~[pinot-all-0.8.0-jar-with-dependencies.jar:0.8.0-c4ceff06d21fc1c1b88469a8dbae742a4b609808]
pinot_1      | 	at org.apache.pinot.sql.parsers.CalciteSqlParser.queryRewrite(CalciteSqlParser.java:395) ~[pinot-all-0.8.0-jar-with-dependencies.jar:0.8.0-c4ceff06d21fc1c1b88469a8dbae742a4b609808]
pinot_1      | 	at org.apache.pinot.sql.parsers.CalciteSqlParser.compileCalciteSqlToPinotQuery(CalciteSqlParser.java:389) ~[pinot-all-0.8.0-jar-with-dependencies.jar:0.8.0-c4ceff06d21fc1c1b88469a8dbae742a4b609808]
pinot_1      | 	at org.apache.pinot.sql.parsers.CalciteSqlParser.compileToPinotQuery(CalciteSqlParser.java:108) ~[pinot-all-0.8.0-jar-with-dependencies.jar:0.8.0-c4ceff06d21fc1c1b88469a8dbae742a4b609808]
pinot_1      | 	at org.apache.pinot.sql.parsers.CalciteSqlCompiler.compileToBrokerRequest(CalciteSqlCompiler.java:35) ~[pinot-all-0.8.0-jar-with-dependencies.jar:0.8.0-c4ceff06d21fc1c1b88469a8dbae742a4b609808]
pinot_1      | 	at org.apache.pinot.controller.api.resources.PinotQueryResource.getQueryResponse(PinotQueryResource.java:166) ~[pinot-all-0.8.0-jar-with-dependencies.jar:0.8.0-c4ceff06d21fc1c1b88469a8dbae742a4b609808]
pinot_1      | 	at org.apache.pinot.controller.api.resources.PinotQueryResource.handlePostSql(PinotQueryResource.java:137) ~[pinot-all-0.8.0-jar-with-dependencies.jar:0.8.0-c4ceff06d21fc1c1b88469a8dbae742a4b609808]
pinot_1      | 	at jdk.internal.reflect.GeneratedMethodAccessor260.invoke(Unknown Source) ~[?:?]
pinot_1      | 	at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:?]
pinot_1      | 	at java.lang.reflect.Method.invoke(Method.java:566) ~[?:?]

@Kamal Chavda I tried it locally, and the exception is thrown when deserializing the geo object

Thanks for checking Jackie.

i think

ST_GeomFromWKB

can convert from regular bytes in wkb

@Yupeng Fu, I tried this

select ST_GeomFromWKB(base64Decode('AQEAACDmEAAAT0wojs27YsA0a4TZX5hOQA==')) from meetupRsvp limit 10

and this

select ST_GeomFromWKB('AQEAACDmEAAAT0wojs27YsA0a4TZX5hOQA==') from meetupRsvp limit 10

both give errors.

what errors do you see?

I get this in the query console:

[
  {
    "message": "QueryExecutionError:\norg.apache.pinot.spi.exception.BadQueryRequestException: Caught exception while initializing transform function: stgeomfromwkb\n\tat org.apache.pinot.core.operator.transform.function.TransformFunctionFactory.get(TransformFunctionFactory.java:207)\n\tat org.apache.pinot.core.operator.transform.TransformOperator.<init>(TransformOperator.java:56)\n\tat org.apache.pinot.core.plan.TransformPlanNode.run(TransformPlanNode.java:56)\n\tat org.apache.pinot.core.plan.SelectionPlanNode.run(SelectionPlanNode.java:83)\n\tat org.apache.pinot.core.plan.CombinePlanNode.run(CombinePlanNode.java:100)\n\tat org.apache.pinot.core.plan.InstanceResponsePlanNode.run(InstanceResponsePlanNode.java:33)\n\tat org.apache.pinot.core.plan.GlobalPlanImplV0.execute(GlobalPlanImplV0.java:45)\n\tat org.apache.pinot.core.query.executor.ServerQueryExecutorV1Impl.processQuery(ServerQueryExecutorV1Impl.java:296)\n\tat org.apache.pinot.core.query.executor.ServerQueryExecutorV1Impl.processQuery(ServerQueryExecutorV1Impl.java:216)\n\tat org.apache.pinot.core.query.executor.QueryExecutor.processQuery(QueryExecutor.java:60)\n\tat org.apache.pinot.core.query.scheduler.QueryScheduler.processQueryAndSerialize(QueryScheduler.java:155)\n\tat org.apache.pinot.core.query.scheduler.QueryScheduler.lambda$createQueryFutureTask$0(QueryScheduler.java:139)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)",
    "errorCode": 200
  }
]

unclear to me. @Jackie any thoughts on debugging?

@Yupeng Fu It doesn't work because we don't have the scalar function version of these functions

hmm why do we need scalar func? i thought this is a transform func in select clause?

transform function cannot be used for ingestion transform. In order to ingest this column, we need to have the function as scalar

ok. but according to the query above

select ST_GeomFromWKB('AQEAACDmEAAAT0wojs27YsA0a4TZX5hOQA==') from meetupRsvp limit 10

, that failed too?

This query (literal only) should be solved on broker side, which requires the scalar function

oh, i see. thats right