Skip to content

Commit d3cec8e

Browse files
sonalgoyalclaude
andauthored
Blocking: build hash schema without spark-mllib SchemaUtils (#1418)
* Blocking: build hash schema without spark-mllib SchemaUtils SparkBlockingTreeUtil.appendHashCol used org.apache.spark.ml.util.SchemaUtils just to append the z_hash column to a schema. That pulls spark-mllib into blocking, and spark-mllib is not on a Spark Connect client classpath, so getBlockHashes failed there with ClassNotFoundException. Use StructType.add with the same duplicate column check SchemaUtils had, so behaviour on classic Spark is unchanged. Adds a unit test for appendHashCol. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Blocking: drop duplicate z_hash column check in appendHashCol Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1 parent cce7be0 commit d3cec8e

2 files changed

Lines changed: 42 additions & 2 deletions

File tree

‎spark/core/src/main/java/zingg/spark/core/util/SparkBlockingTreeUtil.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@
55

66
import org.apache.commons.logging.Log;
77
import org.apache.commons.logging.LogFactory;
8-
import org.apache.spark.ml.util.SchemaUtils;
98
import org.apache.spark.sql.Column;
109
import org.apache.spark.sql.Dataset;
1110
import org.apache.spark.sql.Encoders;
@@ -47,7 +46,9 @@ public ZFrame<Dataset<Row>, Row, Column> getBlockHashes(ZFrame<Dataset<Row>, Row
4746
}
4847

4948
public StructType appendHashCol(StructType s) {
50-
StructType retSchema = SchemaUtils.appendColumn(s, ColName.HASH_COL, DataTypes.IntegerType, false);
49+
// plain StructType instead of spark ml SchemaUtils.appendColumn, so blocking does not need
50+
// spark-mllib on the classpath (it is not there for Spark Connect clients)
51+
StructType retSchema = s.add(ColName.HASH_COL, DataTypes.IntegerType, false);
5152
LOG.debug("returning schema after step 1 is " + retSchema);
5253
return retSchema;
5354
}
Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
package zingg.spark.core.util;
2+
3+
import static org.junit.jupiter.api.Assertions.assertEquals;
4+
import static org.junit.jupiter.api.Assertions.assertFalse;
5+
6+
import org.apache.spark.sql.types.DataTypes;
7+
import org.apache.spark.sql.types.StructField;
8+
import org.apache.spark.sql.types.StructType;
9+
import org.junit.jupiter.api.Test;
10+
11+
import zingg.common.client.util.ColName;
12+
13+
public class TestSparkBlockingTreeUtilSchema {
14+
15+
private final SparkBlockingTreeUtil blockingTreeUtil = new SparkBlockingTreeUtil(null, null);
16+
17+
private StructType getSchema() {
18+
return DataTypes.createStructType(new StructField[] {
19+
DataTypes.createStructField("id", DataTypes.StringType, true),
20+
DataTypes.createStructField("fname", DataTypes.StringType, true)
21+
});
22+
}
23+
24+
@Test
25+
public void testAppendHashColAddsNonNullableIntAtEnd() {
26+
StructType schema = getSchema();
27+
StructType withHash = blockingTreeUtil.appendHashCol(schema);
28+
29+
assertEquals(3, withHash.fields().length);
30+
assertEquals("id", withHash.fields()[0].name());
31+
assertEquals("fname", withHash.fields()[1].name());
32+
StructField hash = withHash.fields()[2];
33+
assertEquals(ColName.HASH_COL, hash.name());
34+
assertEquals(DataTypes.IntegerType, hash.dataType());
35+
assertFalse(hash.nullable());
36+
// input schema is left unchanged
37+
assertEquals(2, schema.fields().length);
38+
}
39+
}

0 commit comments

Comments
 (0)