You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "majian1998 (via GitHub)" <gi...@apache.org> on 2025/11/03 03:38:08 UTC
[PR] Spark 3.5: Support recursive delegate unwrapping to find ExtendedParser in parser chains [iceberg]
majian1998 opened a new pull request, #14483:
URL: https://github.com/apache/iceberg/pull/14483
Related PR: https://github.com/apache/iceberg/pull/13625
Since support for Spark 3.4 will be removed in version 1.11, this fix is only applied to Spark 3.5.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 3.5: Support recursive delegate unwrapping to find ExtendedParser in parser chains [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #14483:
URL: https://github.com/apache/iceberg/pull/14483#issuecomment-3478807357
@majian1998 Is this a clean back-port?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 3.5: Support recursive delegate unwrapping to find ExtendedParser in parser chains [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #14483:
URL: https://github.com/apache/iceberg/pull/14483#issuecomment-3478923223
This is a backport PR; let’s merge it unchanged to keep it identical to the previous one.
@majian1998 Could you open a follow‑up PR (for both 3.5 and 4.0) to address @kamcheungting-db's comments?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 3.5: Support recursive delegate unwrapping to find ExtendedParser in parser chains [iceberg]
Posted by "majian1998 (via GitHub)" <gi...@apache.org>.
majian1998 commented on code in PR #14483:
URL: https://github.com/apache/iceberg/pull/14483#discussion_r2486171753
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/TestExtendedParser.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.iceberg.spark;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Field;
+import java.util.Collections;
+import java.util.List;
+import org.apache.iceberg.NullOrder;
+import org.apache.iceberg.SortDirection;
+import org.apache.iceberg.expressions.Term;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.catalyst.parser.AbstractSqlParser;
+import org.apache.spark.sql.catalyst.parser.AstBuilder;
+import org.apache.spark.sql.catalyst.parser.ParserInterface;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergSparkSqlExtensionsParser;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class TestExtendedParser {
+
+ private static SparkSession spark;
+ private static final String SQL_PARSER_FIELD = "sqlParser";
+ private ParserInterface originalParser;
+
+ @BeforeAll
+ public static void before() {
+ spark = SparkSession.builder().master("local").appName("TestExtendedParser").getOrCreate();
+ }
+
+ @AfterAll
+ public static void after() {
+ if (spark != null) {
+ spark.stop();
+ }
+ }
+
+ @BeforeEach
+ public void saveOriginalParser() throws Exception {
+ Class<?> clazz = spark.sessionState().getClass();
+ Field parserField = null;
+ while (clazz != null && parserField == null) {
+ try {
+ parserField = clazz.getDeclaredField(SQL_PARSER_FIELD);
+ } catch (NoSuchFieldException e) {
+ clazz = clazz.getSuperclass();
+ }
+ }
+ parserField.setAccessible(true);
+ originalParser = (ParserInterface) parserField.get(spark.sessionState());
+ }
+
+ @AfterEach
+ public void restoreOriginalParser() throws Exception {
+ setSessionStateParser(spark.sessionState(), originalParser);
+ }
+
+ /**
+ * Tests that the Iceberg extended SQL parser can correctly parse a sort order string and return
+ * the expected RawOrderField.
+ *
+ * @throws Exception if reflection access fails
+ */
+ @Test
+ public void testParseSortOrderWithRealIcebergExtendedParser() throws Exception {
+ ParserInterface origParser = null;
+ Class<?> clazz = spark.sessionState().getClass();
+ while (clazz != null && origParser == null) {
+ try {
+ Field parserField = clazz.getDeclaredField(SQL_PARSER_FIELD);
+ parserField.setAccessible(true);
+ origParser = (ParserInterface) parserField.get(spark.sessionState());
+ } catch (NoSuchFieldException e) {
+ clazz = clazz.getSuperclass();
+ }
+ }
+ assertThat(origParser).isNotNull();
+
+ IcebergSparkSqlExtensionsParser icebergParser = new IcebergSparkSqlExtensionsParser(origParser);
+
+ setSessionStateParser(spark.sessionState(), icebergParser);
+
+ List<ExtendedParser.RawOrderField> fields =
+ ExtendedParser.parseSortOrder(spark, "id ASC NULLS FIRST");
+
+ assertThat(fields).isNotEmpty();
+ ExtendedParser.RawOrderField first = fields.get(0);
+ assertThat(first.direction()).isEqualTo(SortDirection.ASC);
+ assertThat(first.nullOrder()).isEqualTo(NullOrder.NULLS_FIRST);
+ }
+
+ /**
+ * Tests that parseSortOrder can find and use an ExtendedParser that is wrapped inside another
+ * ParserInterface implementation.
+ *
+ * @throws Exception if reflection access fails
+ */
+ @Test
+ public void testParseSortOrderFindsNestedExtendedParser() throws Exception {
+ ExtendedParser icebergParser = mock(ExtendedParser.class);
+
+ ExtendedParser.RawOrderField field =
+ new ExtendedParser.RawOrderField(
+ mock(Term.class), SortDirection.ASC, NullOrder.NULLS_FIRST);
+ List<ExtendedParser.RawOrderField> expected = Collections.singletonList(field);
+
+ when(icebergParser.parseSortOrder("id ASC NULLS FIRST")).thenReturn(expected);
+
+ ParserInterface wrapper = new WrapperParser(icebergParser);
+
+ setSessionStateParser(spark.sessionState(), wrapper);
+
+ List<ExtendedParser.RawOrderField> result =
+ ExtendedParser.parseSortOrder(spark, "id ASC NULLS FIRST");
+ assertThat(result).isSameAs(expected);
+
+ verify(icebergParser).parseSortOrder("id ASC NULLS FIRST");
Review Comment:
This doesn’t work as expected—verify needs to go after you actually call the method. Otherwise, there’s nothing to verify yet
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 3.5: Support recursive delegate unwrapping to find ExtendedParser in parser chains [iceberg]
Posted by "majian1998 (via GitHub)" <gi...@apache.org>.
majian1998 commented on PR #14483:
URL: https://github.com/apache/iceberg/pull/14483#issuecomment-3478810684
> @majian1998 Is this a clean back-port?
@huaxingao Yes, just a straight copy.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 3.5: Support recursive delegate unwrapping to find ExtendedParser in parser chains [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on PR #14483:
URL: https://github.com/apache/iceberg/pull/14483#issuecomment-3478936906
Merged. Thanks @majian1998 for the PR! Thanks @singhpk234 @kamcheungting-db for the review!
@majian1998 Could you open a follow‑up PR to address @kamcheungting-db's comments?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 3.5: Support recursive delegate unwrapping to find ExtendedParser in parser chains [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao merged PR #14483:
URL: https://github.com/apache/iceberg/pull/14483
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 3.5: Support recursive delegate unwrapping to find ExtendedParser in parser chains [iceberg]
Posted by "kamcheungting-db (via GitHub)" <gi...@apache.org>.
kamcheungting-db commented on code in PR #14483:
URL: https://github.com/apache/iceberg/pull/14483#discussion_r2485298677
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/TestExtendedParser.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.iceberg.spark;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Field;
+import java.util.Collections;
+import java.util.List;
+import org.apache.iceberg.NullOrder;
+import org.apache.iceberg.SortDirection;
+import org.apache.iceberg.expressions.Term;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.catalyst.parser.AbstractSqlParser;
+import org.apache.spark.sql.catalyst.parser.AstBuilder;
+import org.apache.spark.sql.catalyst.parser.ParserInterface;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergSparkSqlExtensionsParser;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class TestExtendedParser {
+
+ private static SparkSession spark;
+ private static final String SQL_PARSER_FIELD = "sqlParser";
+ private ParserInterface originalParser;
+
+ @BeforeAll
+ public static void before() {
+ spark = SparkSession.builder().master("local").appName("TestExtendedParser").getOrCreate();
+ }
+
+ @AfterAll
+ public static void after() {
+ if (spark != null) {
+ spark.stop();
+ }
+ }
+
+ @BeforeEach
+ public void saveOriginalParser() throws Exception {
+ Class<?> clazz = spark.sessionState().getClass();
+ Field parserField = null;
+ while (clazz != null && parserField == null) {
+ try {
+ parserField = clazz.getDeclaredField(SQL_PARSER_FIELD);
+ } catch (NoSuchFieldException e) {
+ clazz = clazz.getSuperclass();
+ }
Review Comment:
add log to help RCA if something goes wrong
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/ExtendedParser.java:
##########
@@ -66,5 +67,43 @@ static List<RawOrderField> parseSortOrder(SparkSession spark, String orderString
}
}
+ private static <T> T findParser(ParserInterface parser, Class<T> clazz) {
+ ParserInterface current = parser;
+ while (current != null) {
+ if (clazz.isInstance(current)) {
+ return clazz.cast(current);
+ }
+
+ ParserInterface next = getNextDelegateParser(current);
+ if (next == null) {
+ break;
+ }
+
+ current = next;
+ }
+
+ return null;
+ }
+
+ private static ParserInterface getNextDelegateParser(ParserInterface parser) {
+ try {
+ Class<?> clazz = parser.getClass();
+ while (clazz != null) {
+ for (Field field : clazz.getDeclaredFields()) {
+ field.setAccessible(true);
+ Object value = field.get(parser);
+ if (value instanceof ParserInterface && value != parser) {
+ return (ParserInterface) value;
+ }
+ }
+ clazz = clazz.getSuperclass();
+ }
+ } catch (Exception e) {
+ // ignore
Review Comment:
shouldn't we at least log something?
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/TestExtendedParser.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.iceberg.spark;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Field;
+import java.util.Collections;
+import java.util.List;
+import org.apache.iceberg.NullOrder;
+import org.apache.iceberg.SortDirection;
+import org.apache.iceberg.expressions.Term;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.catalyst.parser.AbstractSqlParser;
+import org.apache.spark.sql.catalyst.parser.AstBuilder;
+import org.apache.spark.sql.catalyst.parser.ParserInterface;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergSparkSqlExtensionsParser;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class TestExtendedParser {
+
+ private static SparkSession spark;
+ private static final String SQL_PARSER_FIELD = "sqlParser";
+ private ParserInterface originalParser;
+
+ @BeforeAll
+ public static void before() {
+ spark = SparkSession.builder().master("local").appName("TestExtendedParser").getOrCreate();
+ }
+
+ @AfterAll
+ public static void after() {
+ if (spark != null) {
+ spark.stop();
+ }
+ }
+
+ @BeforeEach
+ public void saveOriginalParser() throws Exception {
+ Class<?> clazz = spark.sessionState().getClass();
+ Field parserField = null;
+ while (clazz != null && parserField == null) {
+ try {
+ parserField = clazz.getDeclaredField(SQL_PARSER_FIELD);
+ } catch (NoSuchFieldException e) {
+ clazz = clazz.getSuperclass();
+ }
+ }
+ parserField.setAccessible(true);
+ originalParser = (ParserInterface) parserField.get(spark.sessionState());
+ }
+
+ @AfterEach
+ public void restoreOriginalParser() throws Exception {
+ setSessionStateParser(spark.sessionState(), originalParser);
+ }
+
+ /**
+ * Tests that the Iceberg extended SQL parser can correctly parse a sort order string and return
+ * the expected RawOrderField.
+ *
+ * @throws Exception if reflection access fails
+ */
+ @Test
+ public void testParseSortOrderWithRealIcebergExtendedParser() throws Exception {
+ ParserInterface origParser = null;
+ Class<?> clazz = spark.sessionState().getClass();
+ while (clazz != null && origParser == null) {
+ try {
+ Field parserField = clazz.getDeclaredField(SQL_PARSER_FIELD);
+ parserField.setAccessible(true);
+ origParser = (ParserInterface) parserField.get(spark.sessionState());
+ } catch (NoSuchFieldException e) {
+ clazz = clazz.getSuperclass();
+ }
+ }
+ assertThat(origParser).isNotNull();
+
+ IcebergSparkSqlExtensionsParser icebergParser = new IcebergSparkSqlExtensionsParser(origParser);
+
+ setSessionStateParser(spark.sessionState(), icebergParser);
+
+ List<ExtendedParser.RawOrderField> fields =
+ ExtendedParser.parseSortOrder(spark, "id ASC NULLS FIRST");
+
+ assertThat(fields).isNotEmpty();
+ ExtendedParser.RawOrderField first = fields.get(0);
+ assertThat(first.direction()).isEqualTo(SortDirection.ASC);
+ assertThat(first.nullOrder()).isEqualTo(NullOrder.NULLS_FIRST);
+ }
+
+ /**
+ * Tests that parseSortOrder can find and use an ExtendedParser that is wrapped inside another
+ * ParserInterface implementation.
+ *
+ * @throws Exception if reflection access fails
+ */
+ @Test
+ public void testParseSortOrderFindsNestedExtendedParser() throws Exception {
+ ExtendedParser icebergParser = mock(ExtendedParser.class);
+
+ ExtendedParser.RawOrderField field =
+ new ExtendedParser.RawOrderField(
+ mock(Term.class), SortDirection.ASC, NullOrder.NULLS_FIRST);
+ List<ExtendedParser.RawOrderField> expected = Collections.singletonList(field);
+
+ when(icebergParser.parseSortOrder("id ASC NULLS FIRST")).thenReturn(expected);
+
+ ParserInterface wrapper = new WrapperParser(icebergParser);
+
+ setSessionStateParser(spark.sessionState(), wrapper);
+
+ List<ExtendedParser.RawOrderField> result =
+ ExtendedParser.parseSortOrder(spark, "id ASC NULLS FIRST");
+ assertThat(result).isSameAs(expected);
+
+ verify(icebergParser).parseSortOrder("id ASC NULLS FIRST");
+ }
+
+ /**
+ * Tests that parseSortOrder throws an exception if no ExtendedParser instance can be found in the
+ * parser chain.
+ *
+ * @throws Exception if reflection access fails
+ */
+ @Test
+ public void testParseSortOrderThrowsWhenNoExtendedParserFound() throws Exception {
+ ParserInterface dummy = mock(ParserInterface.class);
+ setSessionStateParser(spark.sessionState(), dummy);
+
+ assertThatThrownBy(() -> ExtendedParser.parseSortOrder(spark, "id ASC"))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("Iceberg ExtendedParser");
+ }
+
+ /**
+ * Tests that parseSortOrder can find an ExtendedParser in a parent class field of the parser.
+ *
+ * @throws Exception if reflection access fails
+ */
+ @Test
+ public void testParseSortOrderFindsExtendedParserInParentClassField() throws Exception {
+ ExtendedParser icebergParser = mock(ExtendedParser.class);
+ ExtendedParser.RawOrderField field =
+ new ExtendedParser.RawOrderField(
+ mock(Term.class), SortDirection.ASC, NullOrder.NULLS_FIRST);
+ List<ExtendedParser.RawOrderField> expected = Collections.singletonList(field);
+ when(icebergParser.parseSortOrder("id ASC NULLS FIRST")).thenReturn(expected);
+ ParserInterface parser = new GrandChildParser(icebergParser);
+ setSessionStateParser(spark.sessionState(), parser);
+
+ List<ExtendedParser.RawOrderField> result =
+ ExtendedParser.parseSortOrder(spark, "id ASC NULLS FIRST");
+ assertThat(result).isSameAs(expected);
+ verify(icebergParser).parseSortOrder("id ASC NULLS FIRST");
Review Comment:
```suggestion
when(icebergParser.parseSortOrder("id ASC NULLS FIRST")).thenReturn(expected);
verify(icebergParser).parseSortOrder("id ASC NULLS FIRST");
ParserInterface parser = new GrandChildParser(icebergParser);
setSessionStateParser(spark.sessionState(), parser);
List<ExtendedParser.RawOrderField> result =
ExtendedParser.parseSortOrder(spark, "id ASC NULLS FIRST");
assertThat(result).isSameAs(expected);
```
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/TestExtendedParser.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.iceberg.spark;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Field;
+import java.util.Collections;
+import java.util.List;
+import org.apache.iceberg.NullOrder;
+import org.apache.iceberg.SortDirection;
+import org.apache.iceberg.expressions.Term;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.catalyst.parser.AbstractSqlParser;
+import org.apache.spark.sql.catalyst.parser.AstBuilder;
+import org.apache.spark.sql.catalyst.parser.ParserInterface;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergSparkSqlExtensionsParser;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class TestExtendedParser {
+
+ private static SparkSession spark;
+ private static final String SQL_PARSER_FIELD = "sqlParser";
+ private ParserInterface originalParser;
+
+ @BeforeAll
+ public static void before() {
+ spark = SparkSession.builder().master("local").appName("TestExtendedParser").getOrCreate();
+ }
+
+ @AfterAll
+ public static void after() {
+ if (spark != null) {
+ spark.stop();
+ }
+ }
+
+ @BeforeEach
+ public void saveOriginalParser() throws Exception {
+ Class<?> clazz = spark.sessionState().getClass();
+ Field parserField = null;
+ while (clazz != null && parserField == null) {
+ try {
+ parserField = clazz.getDeclaredField(SQL_PARSER_FIELD);
+ } catch (NoSuchFieldException e) {
+ clazz = clazz.getSuperclass();
+ }
+ }
+ parserField.setAccessible(true);
+ originalParser = (ParserInterface) parserField.get(spark.sessionState());
+ }
+
+ @AfterEach
+ public void restoreOriginalParser() throws Exception {
+ setSessionStateParser(spark.sessionState(), originalParser);
+ }
+
+ /**
+ * Tests that the Iceberg extended SQL parser can correctly parse a sort order string and return
+ * the expected RawOrderField.
+ *
+ * @throws Exception if reflection access fails
+ */
+ @Test
+ public void testParseSortOrderWithRealIcebergExtendedParser() throws Exception {
+ ParserInterface origParser = null;
+ Class<?> clazz = spark.sessionState().getClass();
+ while (clazz != null && origParser == null) {
+ try {
+ Field parserField = clazz.getDeclaredField(SQL_PARSER_FIELD);
+ parserField.setAccessible(true);
+ origParser = (ParserInterface) parserField.get(spark.sessionState());
+ } catch (NoSuchFieldException e) {
+ clazz = clazz.getSuperclass();
+ }
+ }
+ assertThat(origParser).isNotNull();
+
+ IcebergSparkSqlExtensionsParser icebergParser = new IcebergSparkSqlExtensionsParser(origParser);
+
+ setSessionStateParser(spark.sessionState(), icebergParser);
+
+ List<ExtendedParser.RawOrderField> fields =
+ ExtendedParser.parseSortOrder(spark, "id ASC NULLS FIRST");
+
+ assertThat(fields).isNotEmpty();
+ ExtendedParser.RawOrderField first = fields.get(0);
+ assertThat(first.direction()).isEqualTo(SortDirection.ASC);
+ assertThat(first.nullOrder()).isEqualTo(NullOrder.NULLS_FIRST);
+ }
+
+ /**
+ * Tests that parseSortOrder can find and use an ExtendedParser that is wrapped inside another
+ * ParserInterface implementation.
+ *
+ * @throws Exception if reflection access fails
+ */
+ @Test
+ public void testParseSortOrderFindsNestedExtendedParser() throws Exception {
+ ExtendedParser icebergParser = mock(ExtendedParser.class);
+
+ ExtendedParser.RawOrderField field =
+ new ExtendedParser.RawOrderField(
+ mock(Term.class), SortDirection.ASC, NullOrder.NULLS_FIRST);
+ List<ExtendedParser.RawOrderField> expected = Collections.singletonList(field);
+
+ when(icebergParser.parseSortOrder("id ASC NULLS FIRST")).thenReturn(expected);
+
+ ParserInterface wrapper = new WrapperParser(icebergParser);
+
+ setSessionStateParser(spark.sessionState(), wrapper);
+
+ List<ExtendedParser.RawOrderField> result =
+ ExtendedParser.parseSortOrder(spark, "id ASC NULLS FIRST");
+ assertThat(result).isSameAs(expected);
+
+ verify(icebergParser).parseSortOrder("id ASC NULLS FIRST");
Review Comment:
```suggestion
when(icebergParser.parseSortOrder("id ASC NULLS FIRST")).thenReturn(expected);
verify(icebergParser).parseSortOrder("id ASC NULLS FIRST");
ParserInterface wrapper = new WrapperParser(icebergParser);
setSessionStateParser(spark.sessionState(), wrapper);
List<ExtendedParser.RawOrderField> result =
ExtendedParser.parseSortOrder(spark, "id ASC NULLS FIRST");
assertThat(result).isSameAs(expected);
```
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/ExtendedParser.java:
##########
@@ -66,5 +67,43 @@ static List<RawOrderField> parseSortOrder(SparkSession spark, String orderString
}
}
+ private static <T> T findParser(ParserInterface parser, Class<T> clazz) {
+ ParserInterface current = parser;
+ while (current != null) {
+ if (clazz.isInstance(current)) {
+ return clazz.cast(current);
+ }
+
+ ParserInterface next = getNextDelegateParser(current);
+ if (next == null) {
+ break;
+ }
+
+ current = next;
Review Comment:
```suggestion
current = getNextDelegateParser(current);
```
##########
spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/TestExtendedParser.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.iceberg.spark;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Field;
+import java.util.Collections;
+import java.util.List;
+import org.apache.iceberg.NullOrder;
+import org.apache.iceberg.SortDirection;
+import org.apache.iceberg.expressions.Term;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.catalyst.parser.AbstractSqlParser;
+import org.apache.spark.sql.catalyst.parser.AstBuilder;
+import org.apache.spark.sql.catalyst.parser.ParserInterface;
+import org.apache.spark.sql.catalyst.parser.extensions.IcebergSparkSqlExtensionsParser;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class TestExtendedParser {
+
+ private static SparkSession spark;
+ private static final String SQL_PARSER_FIELD = "sqlParser";
+ private ParserInterface originalParser;
+
+ @BeforeAll
+ public static void before() {
+ spark = SparkSession.builder().master("local").appName("TestExtendedParser").getOrCreate();
+ }
+
+ @AfterAll
+ public static void after() {
+ if (spark != null) {
+ spark.stop();
+ }
+ }
+
+ @BeforeEach
+ public void saveOriginalParser() throws Exception {
+ Class<?> clazz = spark.sessionState().getClass();
+ Field parserField = null;
+ while (clazz != null && parserField == null) {
+ try {
+ parserField = clazz.getDeclaredField(SQL_PARSER_FIELD);
+ } catch (NoSuchFieldException e) {
+ clazz = clazz.getSuperclass();
+ }
+ }
+ parserField.setAccessible(true);
+ originalParser = (ParserInterface) parserField.get(spark.sessionState());
+ }
+
+ @AfterEach
+ public void restoreOriginalParser() throws Exception {
+ setSessionStateParser(spark.sessionState(), originalParser);
+ }
+
+ /**
+ * Tests that the Iceberg extended SQL parser can correctly parse a sort order string and return
+ * the expected RawOrderField.
+ *
+ * @throws Exception if reflection access fails
+ */
+ @Test
+ public void testParseSortOrderWithRealIcebergExtendedParser() throws Exception {
+ ParserInterface origParser = null;
+ Class<?> clazz = spark.sessionState().getClass();
+ while (clazz != null && origParser == null) {
+ try {
+ Field parserField = clazz.getDeclaredField(SQL_PARSER_FIELD);
+ parserField.setAccessible(true);
+ origParser = (ParserInterface) parserField.get(spark.sessionState());
+ } catch (NoSuchFieldException e) {
+ clazz = clazz.getSuperclass();
+ }
Review Comment:
same as above
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 3.5: Support recursive delegate unwrapping to find ExtendedParser in parser chains [iceberg]
Posted by "majian1998 (via GitHub)" <gi...@apache.org>.
majian1998 commented on PR #14483:
URL: https://github.com/apache/iceberg/pull/14483#issuecomment-3478978084
> This is a backport PR; let’s merge it unchanged to keep it identical to the previous one. @majian1998 Could you open a follow‑up PR (for both 3.5 and 4.0) to address @kamcheungting-db's comments?
@huaxingao Got it, I’ll read through the comments and create a follow-up PR for 3.5 and 4.0.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org