Apache SparkのORCとフィルタプッシュダウンのバグを発見してOSSに貢献した話

			
			+---+----+-------------------+
| id|name|              email|
+---+----+-------------------+
|  1|test|               null|
|  2|null|example@example.com|
+---+----+-------------------+

		

			
			df.filter(!(col("name") <=> "dummy"))

		

			
			+---+----+-------------------+
| id|name|              email|
+---+----+-------------------+
|  1|test|               null|
|  2|null|example@example.com|
+---+----+-------------------+

		

			
			+---+----+-------------------+
| id|name|              email|
+---+----+-------------------+
|  1|test|               null|
+---+----+-------------------+

		

			
			import org.apache.spark.sql.functions.col
import org.apache.spark.sql.SparkSession

object NullSafeTest {
  private val projectDir = new java.io.File(".").getCanonicalPath
  private val dataPath = f"$projectDir/data/null_safe_test"

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder
      .appName("NullSafeTest")
      .master("local[*]")
      //      .config("spark.sql.orc.filterPushdown", "false")  // ORC works as expected without filter pushdown
      .getOrCreate()

    import spark.implicits._

    val df = Seq(
      (1, "test", null),
      (2, null, "example@example.com")
    ).toDF("id", "name", "email")

    df.show()

    df.write.mode("overwrite").orc(s"$dataPath/orc")
    df.write.mode("overwrite").parquet(s"$dataPath/parquet")

    val orcDf = spark.read.orc(s"$dataPath/orc")
    val parquetDf = spark.read.parquet(s"$dataPath/parquet")

    val filteredOrcDf = orcDf.filter(!(col("name") <=> "dummy"))
    filteredOrcDf.show()
    filteredOrcDf.explain("extended")

    val filteredParquetDf = parquetDf.filter(!(col("name") <=> "dummy"))
    filteredParquetDf.show()
    filteredParquetDf.explain("extended")

    spark.stop()
  }
}

		

			
			  ORCファイル
  ├── Stripe（ストライプ）
  │   ├── Index Data（インデックスデータ）
  │   ├── Row Data（行データ）
  │   └── Stripe Footer（ストライプフッター）
  ├── Stripe
  │   ├── Index Data
  │   ├── Row Data
  │   └── Stripe Footer
  ├── ...
  ├── File Footer（ファイルフッター）
  └── Postscript（ポストスクリプト）

		

			
			  ストライプ（例: 250MB、100万行）
  │
  ├── Index Data ← ここに「複数の行グループの統計情報」が格納される
  │   ├── 行グループ 0の統計（行 0 ～ 9,999）
  │   ├── 行グループ 1の統計（行 10,000 ～ 19,999）
  │   ├── 行グループ 2の統計（行 20,000 ～ 29,999）
  │   └── ...（行グループ 99まで）
  │
  ├── Row Data ← ここに「実際のデータ全体」が格納される
  │   └── 100万行分のデータ
  │
  └── Stripe Footer

		

			
			  1. ポストスクリプトを読む（ファイルフッターの位置を確認）
     ↓
  2. ファイルフッターを読む
     └── ファイル全体の統計で不要なら終了
     ↓
  3. 必要なストライプのストライプフッターを読む
     └── ストライプの統計で不要ならスキップ
     ↓
  4. インデックスデータを読む（行グループの統計）
     └── 行グループの統計で不要ならスキップ
     ↓
  5. 必要な行データのみ読む
     └── 必要なカラムのみ読む

		

			
			  static TruthValue evaluatePredicateRange(PredicateLeaf predicate,
                                            ValueRange range,
                                            BloomFilter bloomFilter,
                                            boolean useUTCTimestamp) {

    // 省略

    if (!range.hasValues()) {
      if (predicate.getOperator() == PredicateLeaf.Operator.IS_NULL) {
        return TruthValue.YES;
+     } else if (predicate.getOperator() == PredicateLeaf.Operator.NULL_SAFE_EQUALS) {
+       Object literal = predicate.getLiteral();
+       if (literal == null) {
+         return TruthValue.YES;
+       } else {
+         return TruthValue.NO;
+       }
      } else {
        return TruthValue.NULL;
      }
    } else if (!range.isComparable()) {
      return range.hasNulls ? TruthValue.YES_NO_NULL : TruthValue.YES_NO;
    }

    // 省略

  }

		

			
			static class ValueRange<T extends Comparable> {
    final Comparable lower;        // 最小値
    final Comparable upper;        // 最大値
    final boolean onlyLowerBound;  // 下限のみが提供されているか
    final boolean onlyUpperBound;  // 上限のみが提供されているか
    final boolean hasNulls;        // NULL値が含まれているか
    final boolean hasValue;        // 値が存在するか
    final boolean comparable;      // 比較可能か
}

		

			
			+---+----+-------------------+
| id|name|              email|
+---+----+-------------------+
|  1|null|hoge001@example.com|
|  2|null|hoge002@example.com|
|  3|null|hoge003@example.com|
+---+----+-------------------+

		

			
			/**
  * 各行グループに対してフィルタ（述語）プッシュダウンを評価し、
  * 読み込むべき行グループを判定する
  * 
  * @param stripe ストライプ情報
  * @param indexes 行グループごとの統計情報インデックス
  * @param bloomFilterKinds ブルームフィルタの種類
  * @param encodings カラムのエンコーディング情報
  * @param bloomFilterIndices ブルームフィルタインデックス
  * @param returnNone 全行グループがスキップされた場合にREAD_NO_RGSを返すか
  * @return 各行グループを読み込むかどうかのboolean配列
  */
public boolean[] pickRowGroups(StripeInformation stripe,
                                OrcProto.RowIndex[] indexes,
                                OrcProto.Stream.Kind[] bloomFilterKinds,
                                List<OrcProto.ColumnEncoding> encodings,
                                OrcProto.BloomFilterIndex[] bloomFilterIndices,
                                boolean returnNone) throws IOException {

  // 結果配列: true = 行グループを読み込む, false = スキップ
  boolean[] result = new boolean[groupsInStripe];

  // 各述語リーフの評価結果を格納する配列
  TruthValue[] leafValues = new TruthValue[sargLeaves.size()];

  // 各行グループに対して述語を評価
  for (int rowGroup = 0; rowGroup < result.length; ++rowGroup) {
    for (int pred = 0; pred < leafValues.length; ++pred) {
      int columnIx = filterColumns[pred];

      // 行グループの統計情報を取得
      OrcProto.RowIndexEntry entry = indexes[columnIx].getEntry(rowGroup);
      OrcProto.ColumnStatistics stats = entry.getStatistics();

      PredicateLeaf predicate = sargLeaves.get(pred);

      // ★統計情報に基づいて述語を評価★
      // この中でevaluatePredicate()が呼ばれ、修正箇所(lines 766-772)が実行される
      leafValues[pred] = evaluatePredicateProto(stats, predicate, ...);
    }

    // ★述語式全体を評価して行グループの読み込み可否を判定★
    // sarg.evaluate()で述語式全体(AND/ORの組み合わせ)を評価
    // isNeeded()で最終的な読み込み可否を判定
    // - TruthValue.YES, YES_NO, YES_NULL, YES_NO_NULL → true (読み込む)
    // - TruthValue.NO, NULL, NO_NULL → false (スキップ)
    result[rowGroup] = sarg.evaluate(leafValues).isNeeded();
  }

  return result;
}

		

			
			df.filter(!(col("name") <=> "dummy"))

		

			
			+---+----+-------------------+
| id|name|              email|
+---+----+-------------------+
|  1|test|               null|
|  2|null|example@example.com|
+---+----+-------------------+

		

			
			% sysctl -n hw.physicalcpu_max
10

		

			
			part-00000-xxx.orc (行1のデータ)
└── ストライプ 0: 1行
    └── 行グループ 0: 1行
        └── nameカラムの統計情報: hasValue: true, hasNulls: false, min: test, max: test

part-00001-xxx.orc (行2のデータ)
└── ストライプ 0: 1行
    └── 行グループ 0: 1行
        └── nameカラムの統計情報: hasValue: false, hasNulls: true

		

			
			df.repartition(1).write.mode("overwrite").orc(s"$dataPath/orc")