optimize truncate(col) == value to startsWith predicate

Rewrite truncate(col, width) == "value" predicates to col STARTS_WITH "value"
for string columns when literal.length() == width. This enables better predicate
pushdown to storage formats and efficient use of prefix indexes.

The optimization only applies when ALL conditions are met:
- Operation is equality
- Term is a truncate transform
- Literal is a string type
- Literal length equals truncate width (CRITICAL for correctness)

Without the width check, truncate(col, 10) == "abc" would incorrectly match
"abc1234567" if rewritten to STARTS_WITH.

Added comprehensive tests including width mismatch edge case.

Author: shangxinli

src/iceberg/expression/predicate.cc

@@ -21,11 +21,14 @@
 
 #include <algorithm>
 #include <format>
+#include <regex>
 
 #include "iceberg/exception.h"
 #include "iceberg/expression/expressions.h"
 #include "iceberg/expression/literal.h"
+#include "iceberg/expression/term.h"
 #include "iceberg/result.h"
+#include "iceberg/transform.h"
 #include "iceberg/type.h"
 #include "iceberg/util/checked_cast.h"
 #include "iceberg/util/formatter_internal.h"
@@ -226,7 +229,37 @@ Result<std::shared_ptr<Expression>> UnboundPredicate<B>::BindLiteralOperation(
     }
   }
 
-  // TODO(gangwu): translate truncate(col) == value to startsWith(value)
+  // Optimize: translate truncate(col, width) == value to col startsWith(value)
+  // This optimization allows better predicate pushdown and index usage
+  // IMPORTANT: Only valid when literal.length() == width
+  if (BASE::op() == Expression::Operation::kEq &&
+      bound_term->kind() == Term::Kind::kTransform) {
+    auto* transform_term = dynamic_cast<BoundTransform*>(bound_term.get());
+    if (transform_term &&
+        transform_term->transform()->transform_type() == TransformType::kTruncate &&
+        literal.type()->type_id() == TypeId::kString) {
+      // Extract width from transform string (format: "truncate[width]")
+      std::string transform_str = transform_term->transform()->ToString();
+      std::regex width_regex(R"(truncate\[(\d+)\])");
+      std::smatch match;
+
+      if (std::regex_match(transform_str, match, width_regex)) {
+        int32_t truncate_width = std::stoi(match[1].str());
+        auto& string_value = std::get<std::string>(literal.value());
+
+        // Only optimize if literal length equals truncate width
+        // Example: truncate(col, 5) == "Alice" can be optimized
+        //          truncate(col, 10) == "abc" CANNOT (would incorrectly match "abc1234567")
+        if (static_cast<int32_t>(string_value.length()) == truncate_width) {
+          // Rewrite: truncate(col, width) == "value" → col STARTS_WITH "value"
+          return std::make_shared<BoundLiteralPredicate>(
+              Expression::Operation::kStartsWith, transform_term->reference(),
+              std::move(literal));
+        }
+      }
+    }
+  }
+
   return std::make_shared<BoundLiteralPredicate>(BASE::op(), std::move(bound_term),
                                                  std::move(literal));
 }

src/iceberg/test/predicate_test.cc

@@ -18,6 +18,7 @@
  */
 
 #include "iceberg/expression/expressions.h"
+#include "iceberg/expression/predicate.h"
 #include "iceberg/schema.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/type.h"
@@ -433,4 +434,127 @@ TEST_F(PredicateTest, ComplexExpressionCombinations) {
   EXPECT_EQ(nested->op(), Expression::Operation::kAnd);
 }
 
+TEST_F(PredicateTest, TruncateOptimizationToStartsWith) {
+  // Test that truncate(col) == "value" is optimized to col STARTS_WITH "value"
+
+  // Create a truncate transform expression: truncate(name, 5)
+  auto truncate_expr = Expressions::Truncate("name", 5);
+
+  // Create predicate: truncate(name, 5) == "Alice"
+  auto truncate_eq_pred = std::make_shared<UnboundPredicate<BoundTransform>>(
+      Expression::Operation::kEq, truncate_expr, Literal::String("Alice"));
+
+  // Bind the predicate to the schema
+  auto bound_result = truncate_eq_pred->Bind(*schema_, /*case_sensitive=*/true);
+  ASSERT_THAT(bound_result, IsOk());
+  auto bound_pred = bound_result.value();
+
+  // After optimization, it should be a STARTS_WITH operation
+  EXPECT_EQ(bound_pred->op(), Expression::Operation::kStartsWith);
+
+  // Verify it's a BoundLiteralPredicate
+  auto* literal_pred = dynamic_cast<BoundLiteralPredicate*>(bound_pred.get());
+  ASSERT_NE(literal_pred, nullptr);
+
+  // The term should now be a direct reference to "name", not a transform
+  EXPECT_EQ(literal_pred->term()->kind(), Term::Kind::kReference);
+
+  // The literal should still be "Alice"
+  EXPECT_EQ(literal_pred->literal(), Literal::String("Alice"));
+}
+
+TEST_F(PredicateTest, TruncateOptimizationNotAppliedForNonEquality) {
+  // Test that optimization is NOT applied for non-equality operations
+
+  auto truncate_expr = Expressions::Truncate("name", 5);
+
+  // Test with less-than (should NOT be optimized)
+  auto truncate_lt_pred = std::make_shared<UnboundPredicate<BoundTransform>>(
+      Expression::Operation::kLt, truncate_expr, Literal::String("Bob"));
+  auto bound_lt_result = truncate_lt_pred->Bind(*schema_, /*case_sensitive=*/true);
+  ASSERT_THAT(bound_lt_result, IsOk());
+  auto bound_lt = bound_lt_result.value();
+
+  // Should remain as kLt, not converted to STARTS_WITH
+  EXPECT_EQ(bound_lt->op(), Expression::Operation::kLt);
+
+  // The term should still be a transform
+  auto* literal_pred_lt = dynamic_cast<BoundLiteralPredicate*>(bound_lt.get());
+  ASSERT_NE(literal_pred_lt, nullptr);
+  EXPECT_EQ(literal_pred_lt->term()->kind(), Term::Kind::kTransform);
+}
+
+TEST_F(PredicateTest, TruncateOptimizationNotAppliedForNonString) {
+  // Test that optimization is NOT applied for non-string types
+  // (truncate can also work on binary types, but optimization only applies to strings)
+
+  // Create a schema with binary field
+  auto binary_schema = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeOptional(1, "data", binary())},
+      /*schema_id=*/0);
+
+  auto truncate_expr = Expressions::Truncate("data", 10);
+  auto truncate_eq_pred = std::make_shared<UnboundPredicate<BoundTransform>>(
+      Expression::Operation::kEq, truncate_expr,
+      Literal::Binary({0x01, 0x02, 0x03, 0x04, 0x05}));
+
+  auto bound_result = truncate_eq_pred->Bind(*binary_schema, /*case_sensitive=*/true);
+  ASSERT_THAT(bound_result, IsOk());
+  auto bound_pred = bound_result.value();
+
+  // Should remain as kEq, not converted to STARTS_WITH (binary doesn't support startsWith)
+  EXPECT_EQ(bound_pred->op(), Expression::Operation::kEq);
+
+  // The term should still be a transform
+  auto* literal_pred = dynamic_cast<BoundLiteralPredicate*>(bound_pred.get());
+  ASSERT_NE(literal_pred, nullptr);
+  EXPECT_EQ(literal_pred->term()->kind(), Term::Kind::kTransform);
+}
+
+TEST_F(PredicateTest, TruncateOptimizationNotAppliedForWidthMismatch) {
+  // CRITICAL TEST: Optimization must NOT apply when literal length != truncate width
+  // Example: truncate(col, 10) == "abc" should NOT become STARTS_WITH
+  // Because "abc1234567" would match STARTS_WITH but NOT truncate equality
+
+  auto truncate_expr = Expressions::Truncate("name", 10);
+
+  // Literal "abc" has length 3, but truncate width is 10
+  auto truncate_eq_pred = std::make_shared<UnboundPredicate<BoundTransform>>(
+      Expression::Operation::kEq, truncate_expr, Literal::String("abc"));
+
+  auto bound_result = truncate_eq_pred->Bind(*schema_, /*case_sensitive=*/true);
+  ASSERT_THAT(bound_result, IsOk());
+  auto bound_pred = bound_result.value();
+
+  // Should remain as kEq, NOT converted to STARTS_WITH
+  EXPECT_EQ(bound_pred->op(), Expression::Operation::kEq);
+
+  // The term should still be a transform (not optimized away)
+  auto* literal_pred = dynamic_cast<BoundLiteralPredicate*>(bound_pred.get());
+  ASSERT_NE(literal_pred, nullptr);
+  EXPECT_EQ(literal_pred->term()->kind(), Term::Kind::kTransform);
+}
+
+TEST_F(PredicateTest, TruncateOptimizationAppliedWhenLengthMatches) {
+  // Test that optimization IS applied when literal length == truncate width
+
+  auto truncate_expr = Expressions::Truncate("name", 5);
+
+  // Literal "Alice" has length 5, matching truncate width 5
+  auto truncate_eq_pred = std::make_shared<UnboundPredicate<BoundTransform>>(
+      Expression::Operation::kEq, truncate_expr, Literal::String("Alice"));
+
+  auto bound_result = truncate_eq_pred->Bind(*schema_, /*case_sensitive=*/true);
+  ASSERT_THAT(bound_result, IsOk());
+  auto bound_pred = bound_result.value();
+
+  // Should be optimized to STARTS_WITH
+  EXPECT_EQ(bound_pred->op(), Expression::Operation::kStartsWith);
+
+  // The term should be a direct reference (optimization applied)
+  auto* literal_pred = dynamic_cast<BoundLiteralPredicate*>(bound_pred.get());
+  ASSERT_NE(literal_pred, nullptr);
+  EXPECT_EQ(literal_pred->term()->kind(), Term::Kind::kReference);
+}
+
 }  // namespace iceberg