Convert error while query imported spark table

[yujiantao]

We use SparkTableUtil.importSparkTable to create an iceberg table on parquet data files. The iceberg table schema is built from spark table schema, and the metrics is built from parquet schema and persisted in manifest file. If the two schemas are not same, when we query the table, the id of the filtered field in iceberg table schema is used to get upper bound from metrics, then we get wrong upper bound or some converting error.

Legacy parquet files may not have consistent schema, so we should not use schema read from parquet footer to build the fleld id in metrics.

Exception in thread "main" org.apache.iceberg.exceptions.RuntimeIOException: Failed to decode value as UTF-8: java.nio.HeapByteBuffer[pos=0 lim=8 cap=8]
	at org.apache.iceberg.types.Conversions.internalFromByteBuffer(Conversions.java:170)
	at org.apache.iceberg.types.Conversions.fromByteBuffer(Conversions.java:130)
	at org.apache.iceberg.expressions.InclusiveMetricsEvaluator$MetricsEvalVisitor.eq(InclusiveMetricsEvaluator.java:242)
	at org.apache.iceberg.expressions.InclusiveMetricsEvaluator$MetricsEvalVisitor.eq(InclusiveMetricsEvaluator.java:83)
	at org.apache.iceberg.expressions.ExpressionVisitors$BoundExpressionVisitor.predicate(ExpressionVisitors.java:120)
	at org.apache.iceberg.expressions.ExpressionVisitors.visitEvaluator(ExpressionVisitors.java:195)
	at org.apache.iceberg.expressions.ExpressionVisitors.visitEvaluator(ExpressionVisitors.java:210)
	at org.apache.iceberg.expressions.InclusiveMetricsEvaluator$MetricsEvalVisitor.eval(InclusiveMetricsEvaluator.java:99)
	at org.apache.iceberg.expressions.InclusiveMetricsEvaluator$MetricsEvalVisitor.access$000(InclusiveMetricsEvaluator.java:83)
	at org.apache.iceberg.expressions.InclusiveMetricsEvaluator.eval(InclusiveMetricsEvaluator.java:77)
	at org.apache.iceberg.FilteredManifest.lambda$iterator$3(FilteredManifest.java:144)
	at com.google.common.collect.Iterators$5.computeNext(Iterators.java:637)
	at com.google.common.collect.AbstractIterator.tryToComputeNext(AbstractIterator.java:141)
	at com.google.common.collect.AbstractIterator.hasNext(AbstractIterator.java:136)
	at com.google.common.collect.TransformedIterator.hasNext(TransformedIterator.java:42)
	at org.apache.iceberg.io.CloseableIterable$3$1.hasNext(CloseableIterable.java:89)
	at org.apache.iceberg.io.CloseableIterable$ConcatCloseableIterable$ConcatCloseableIterator.hasNext(CloseableIterable.java:142)
	at com.google.common.collect.TransformedIterator.hasNext(TransformedIterator.java:42)
	at com.google.common.collect.TransformedIterator.hasNext(TransformedIterator.java:42)
	at com.google.common.collect.Iterators$ConcatenatedIterator.getTopMetaIterator(Iterators.java:1309)
	at com.google.common.collect.Iterators$ConcatenatedIterator.hasNext(Iterators.java:1325)
	at org.apache.iceberg.util.BinPacking$PackingIterator.hasNext(BinPacking.java:101)
	at org.apache.iceberg.io.CloseableIterable$3$1.hasNext(CloseableIterable.java:89)
	at com.google.common.collect.Iterators.addAll(Iterators.java:355)
	at com.google.common.collect.Lists.newArrayList(Lists.java:143)
	at com.google.common.collect.Lists.newArrayList(Lists.java:130)
	at org.apache.iceberg.spark.source.Reader.tasks(Reader.java:265)
	at org.apache.iceberg.spark.source.Reader.planInputPartitions(Reader.java:167)
	at org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanExec.partitions$lzycompute(DataSourceV2ScanExec.scala:76)
	at org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanExec.partitions(DataSourceV2ScanExec.scala:75)
	at org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanExec.outputPartitioning(DataSourceV2ScanExec.scala:65)
	at org.apache.spark.sql.execution.exchange.EnsureRequirements$$anonfun$org$apache$spark$sql$execution$exchange$EnsureRequirements$$ensureDistributionAndOrdering$1.apply(EnsureRequirements.scala:150)
	at org.apache.spark.sql.execution.exchange.EnsureRequirements$$anonfun$org$apache$spark$sql$execution$exchange$EnsureRequirements$$ensureDistributionAndOrdering$1.apply(EnsureRequirements.scala:149)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	at scala.collection.immutable.List.foreach(List.scala:392)
	at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
	at scala.collection.immutable.List.map(List.scala:296)
	at org.apache.spark.sql.execution.exchange.EnsureRequirements.org$apache$spark$sql$execution$exchange$EnsureRequirements$$ensureDistributionAndOrdering(EnsureRequirements.scala:149)
	at org.apache.spark.sql.execution.exchange.EnsureRequirements$$anonfun$apply$1.applyOrElse(EnsureRequirements.scala:304)
	at org.apache.spark.sql.execution.exchange.EnsureRequirements$$anonfun$apply$1.applyOrElse(EnsureRequirements.scala:296)
	at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformUp$1.apply(TreeNode.scala:278)
	at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformUp$1.apply(TreeNode.scala:278)
	at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:70)
	at org.apache.spark.sql.catalyst.trees.TreeNode.transformUp(TreeNode.scala:277)
	at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$3.apply(TreeNode.scala:275)
	at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$3.apply(TreeNode.scala:275)
	at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:326)
	at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:187)
	at org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:324)
	at org.apache.spark.sql.catalyst.trees.TreeNode.transformUp(TreeNode.scala:275)
	at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$3.apply(TreeNode.scala:275)
	at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$3.apply(TreeNode.scala:275)
	at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:326)
	at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:187)
	at org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:324)
	at org.apache.spark.sql.catalyst.trees.TreeNode.transformUp(TreeNode.scala:275)
	at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$3.apply(TreeNode.scala:275)
	at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$3.apply(TreeNode.scala:275)
	at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:326)
	at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:187)
	at org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:324)
	at org.apache.spark.sql.catalyst.trees.TreeNode.transformUp(TreeNode.scala:275)
	at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$3.apply(TreeNode.scala:275)
	at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$3.apply(TreeNode.scala:275)
	at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:326)
	at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:187)
	at org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:324)
	at org.apache.spark.sql.catalyst.trees.TreeNode.transformUp(TreeNode.scala:275)
	at org.apache.spark.sql.execution.exchange.EnsureRequirements.apply(EnsureRequirements.scala:296)
	at org.apache.spark.sql.execution.exchange.EnsureRequirements.apply(EnsureRequirements.scala:38)
	at org.apache.spark.sql.execution.QueryExecution$$anonfun$prepareForExecution$1.apply(QueryExecution.scala:87)
	at org.apache.spark.sql.execution.QueryExecution$$anonfun$prepareForExecution$1.apply(QueryExecution.scala:87)
	at scala.collection.LinearSeqOptimized$class.foldLeft(LinearSeqOptimized.scala:124)
	at scala.collection.immutable.List.foldLeft(List.scala:84)
	at org.apache.spark.sql.execution.QueryExecution.prepareForExecution(QueryExecution.scala:87)
	at org.apache.spark.sql.execution.QueryExecution.executedPlan$lzycompute(QueryExecution.scala:77)
	at org.apache.spark.sql.execution.QueryExecution.executedPlan(QueryExecution.scala:77)
	at org.apache.spark.sql.execution.QueryExecution$$anonfun$toString$3.apply(QueryExecution.scala:207)
	at org.apache.spark.sql.execution.QueryExecution$$anonfun$toString$3.apply(QueryExecution.scala:207)
	at org.apache.spark.sql.execution.QueryExecution.stringOrError(QueryExecution.scala:99)
	at org.apache.spark.sql.execution.QueryExecution.toString(QueryExecution.scala:207)
	at org.apache.spark.sql.execution.SQLExecution$$anonfun$withNewExecutionId$1.apply(SQLExecution.scala:75)
	at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:125)
	at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:73)
	at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:676)
	at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:285)
	at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:271)
	at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:229)
	at com.ushareit.data.iceberg.poc.TestSingleCTRIceberg$.main(TestSingleCTRIceberg.scala:111)
	at com.ushareit.data.iceberg.poc.TestSingleCTRIceberg.main(TestSingleCTRIceberg.scala)
Caused by: java.nio.charset.MalformedInputException: Input length = 2
	at java.nio.charset.CoderResult.throwException(CoderResult.java:281)
	at java.nio.charset.CharsetDecoder.decode(CharsetDecoder.java:816)
	at org.apache.iceberg.types.Conversions.internalFromByteBuffer(Conversions.java:168)
	... 90 more

[rdblue]

@yujiantao, the current fallback strategy for assigning IDs assumes that your data has not had any columns removed. We've talked about adding support for NameMapping to Parquet so that it will fall back to name-based resolution. We've built this for Avro since Avro only supports name-based resolution (Parquet supports both name-based and position-based in some systems).

Would you like to work on adding an ID assignment fallback implementation that uses NameMapping?

[chenjunjiedada]

@rdblue, Do we still need to add this base on rencent NameMapping changes? If there is no on going effort on this, I 'd like to take this to get familiar with schema things.

[rdblue]

@chenjunjiedada, I think we could use some work here if you're interested.

The basic problem is what to do when there are data files without IDs. Right now, we have a fallback strategy for Parquet that uses column position because we use position-based resolution in our tables. But another valid strategy is to use a NameMapping for people that use name-based column resolution. Implementing the name-based resolution for Parquet is a good project.

[chenjunjiedada]

@rdblue , thanks for the explanation, I will take this.