ColumnarBatchScanExec reading parquet failed on java.lang.IllegalArgumentException: not all nodes and buffers were consumed

[xuechendi]

Describe the bug
Failed to do aggregate when running nativesql plugin with jupyter notebook and enabled ColumnarBatchScan
on java.lang.IllegalArgumentException: not all nodes and buffers were consumed

To Reproduce

native_sql_path = "/mnt/nvme2/chendi/intel-bigdata/OAP/native-sql-engine/native-sql-engine/core/target/spark-columnar-core-1.2.0-snapshot-jar-with-dependencies.jar"
native_arrow_datasource_path = "/mnt/nvme2/chendi/intel-bigdata/OAP/native-sql-engine/arrow-data-source/standard/target/spark-arrow-datasource-standard-1.2.0-snapshot-jar-with-dependencies.jar"

spark = SparkSession.builder.master('yarn')\
        .appName("Recsys2021_data_process")\
        .config("spark.memory.offHeap.size", "16G")\
        .config("spark.executorEnv.LD_LIBRARY_PATH", "/usr/local/lib64/")\
        .config("spark.driver.extraClassPath", 
                f"{native_sql_path}:{native_arrow_datasource_path}")\
        .config("spark.executor.extraClassPath",
                f"{native_sql_path}:{native_arrow_datasource_path}")\
        .config("spark.sql.extensions", "com.intel.oap.ColumnarPlugin")\
        .config("spark.shuffle.manager", "org.apache.spark.shuffle.sort.ColumnarShuffleManager")\
        .config("spark.oap.sql.columnar.batchscan", "true")\
        .getOrCreate()

df = spark.read.format("arrow").load("/recsys2021_0608")
print(df.count())

Additional context

Caused by: java.lang.IllegalArgumentException: not all nodes and buffers were consumed. nodes: [ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=10000, nullCount=7927], ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=10000, nullCount=6195], ArrowFieldNode [length=10000, nullCount=8768], ArrowFieldNode [length=10000, nullCount=8768], ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=10000, nullCount=9742], ArrowFieldNode [length=10000, nullCount=9167], ArrowFieldNode [length=10000, nullCount=9932], ArrowFieldNode [length=10000, nullCount=5993], ArrowFieldNode [length=10000, nullCount=0], ArrowFieldNode [length=472721, nullCount=0]] buffers: [ArrowBuf[1854], address:0, length:0, ArrowBuf[1855], address:139851195947392, length:40004, ArrowBuf[1856], address:139850638208640, length:2535645, ArrowBuf[1857], address:139836344241152, length:1250, ArrowBuf[1858], address:139851196090432, length:40004, ArrowBuf[1859], address:139850561570176, length:137319, ArrowBuf[1860], address:0, length:0, ArrowBuf[1861], address:139850532177152, length:40004, ArrowBuf[1862], address:139850644515520, length:320000, ArrowBuf[1863], address:139836344239872, length:1250, ArrowBuf[1864], address:139850562265792, length:40004, ArrowBuf[1865], address:139850532222336, length:28971, ArrowBuf[1866], address:139836344238592, length:1250, ArrowBuf[1867], address:139851195759296, length:40004, ArrowBuf[1868], address:139850645416192, length:41239, ArrowBuf[1869], address:139836344237312, length:1250, ArrowBuf[1870], address:139850531473344, length:40004, ArrowBuf[1871], address:139850645469056, length:41239, ArrowBuf[1872], address:0, length:0, ArrowBuf[1873], address:139850531559744, length:40004, ArrowBuf[1874], address:139850796542976, length:74222, ArrowBuf[1875], address:0, length:0, ArrowBuf[1876], address:139850545969024, length:40004, ArrowBuf[1877], address:139850793378560, length:320000, ArrowBuf[1878], address:139836344234752, length:1250, ArrowBuf[1879], address:139850793711104, length:80000, ArrowBuf[1880], address:0, length:0, ArrowBuf[1881], address:139850793796800, length:40004, ArrowBuf[1882], address:139850795999296, length:320000, ArrowBuf[1883], address:139836344233472, length:1250, ArrowBuf[1884], address:139850795750912, length:80000, ArrowBuf[1885], address:139836344232192, length:1250, ArrowBuf[1886], address:139850796333504, length:80000, ArrowBuf[1887], address:0, length:0, ArrowBuf[1888], address:139836344230912, length:1250, ArrowBuf[1889], address:139836344229632, length:1250, ArrowBuf[1890], address:139850802610688, length:80000, ArrowBuf[1891], address:0, length:0, ArrowBuf[1892], address:139850546106048, length:40004, ArrowBuf[1893], address:139850810514752, length:320000, ArrowBuf[1894], address:139836344228352, length:1250, ArrowBuf[1895], address:139850805187008, length:80000, ArrowBuf[1896], address:139836344227072, length:1250, ArrowBuf[1897], address:139850805276416, length:80000, ArrowBuf[1898], address:0, length:0, ArrowBuf[1899], address:139836344225792, length:1250, ArrowBuf[1900], address:139841170389760, length:1250, ArrowBuf[1901], address:139850811083456, length:80000, ArrowBuf[1902], address:0, length:0, ArrowBuf[1903], address:139841170388480, length:1250, ArrowBuf[1904], address:139841170387200, length:1250, ArrowBuf[1905], address:139850814475200, length:80000, ArrowBuf[1906], address:139841170385920, length:1250, ArrowBuf[1907], address:139850814611264, length:80000, ArrowBuf[1908], address:139841170384640, length:1250, ArrowBuf[1909], address:139850814744832, length:80000, ArrowBuf[1910], address:139841170383360, length:1250, ArrowBuf[1911], address:139850836380416, length:80000, ArrowBuf[1912], address:0, length:0, ArrowBuf[1913], address:139850814560192, length:40004, ArrowBuf[1914], address:139850836467456, length:59091, ArrowBuf[1915], address:139850858430272, length:3781768]
	at org.apache.arrow.vector.VectorLoader.load(VectorLoader.java:86)
	at com.intel.oap.vectorized.ArrowWritableColumnVector.loadColumns(ArrowWritableColumnVector.java:137)
	at com.intel.oap.vectorized.ArrowWritableColumnVector.loadColumns(ArrowWritableColumnVector.java:127)
	at com.intel.oap.spark.sql.execution.datasources.v2.arrow.ArrowUtils$.loadBatch(ArrowUtils.scala:96)
	at com.intel.oap.spark.sql.execution.datasources.v2.arrow.ArrowPartitionReaderFactory.$anonfun$buildColumnarReader$4(ArrowPartitionReaderFactory.scala:85)
	at scala.collection.Iterator$$anon$10.next(Iterator.scala:459)
	at com.intel.oap.spark.sql.execution.datasources.v2.arrow.ArrowPartitionReaderFactory$$anon$1.get(ArrowPartitionReaderFactory.scala:96)
	at com.intel.oap.spark.sql.execution.datasources.v2.arrow.ArrowPartitionReaderFactory$$anon$1.get(ArrowPartitionReaderFactory.scala:87)
	at org.apache.spark.sql.execution.datasources.v2.PartitionedFileReader.get(FilePartitionReaderFactory.scala:56)
	at org.apache.spark.sql.execution.datasources.v2.FilePartitionReader.get(FilePartitionReader.scala:96)
	at com.intel.oap.execution.ColumnarDataSourceRDD$$anon$1.next(ColumnarDataSourceRDD.scala:110)
	at com.intel.oap.vectorized.CloseableColumnBatchIterator.next(CloseableColumnBatchIterator.scala:52)
	at com.intel.oap.vectorized.CloseableColumnBatchIterator.next(CloseableColumnBatchIterator.scala:29)
	at org.apache.spark.InterruptibleIterator.next(InterruptibleIterator.scala:40)
	at scala.collection.Iterator$$anon$10.next(Iterator.scala:459)
	at com.intel.oap.execution.ColumnarHashAggregateExec$$anon$1.process(ColumnarHashAggregateExec.scala:163)
	at com.intel.oap.execution.ColumnarHashAggregateExec$$anon$1.hasNext(ColumnarHashAggregateExec.scala:206)
	at com.intel.oap.vectorized.CloseableColumnBatchIterator.hasNext(CloseableColumnBatchIterator.scala:47)
	at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
	at org.apache.spark.shuffle.ColumnarShuffleWriter.write(ColumnarShuffleWriter.scala:84)
	at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)
	at org.apache.spark.scheduler.Task.run(Task.scala:131)
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:497)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1439)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:500)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

[xuechendi]

@zhztheplayer , pls take a look