GH-51649: [C++][Parquet] Tell encoding fuzzer to skip inputs with invalid parameters

cpp/src/arrow/csv/fuzz.cc

@@ -121,6 +121,5 @@ Status FuzzCsvReader(const uint8_t* data, int64_t size) {
 
 extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
   auto status = arrow::csv::FuzzCsvReader(data, static_cast<int64_t>(size));
-  arrow::internal::LogFuzzStatus(status, data, size);
-  return 0;
+  return arrow::internal::LogFuzzStatus(status, data, size);
 }

cpp/src/arrow/ipc/file_fuzz.cc

@@ -24,6 +24,5 @@
 
 extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
   auto status = arrow::ipc::internal::FuzzIpcFile(data, static_cast<int64_t>(size));
-  arrow::internal::LogFuzzStatus(status, data, static_cast<int64_t>(size));
-  return 0;
+  return arrow::internal::LogFuzzStatus(status, data, static_cast<int64_t>(size));
 }

cpp/src/arrow/ipc/stream_fuzz.cc

@@ -24,6 +24,5 @@
 
 extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
   auto status = arrow::ipc::internal::FuzzIpcStream(data, static_cast<int64_t>(size));
-  arrow::internal::LogFuzzStatus(status, data, static_cast<int64_t>(size));
-  return 0;
+  return arrow::internal::LogFuzzStatus(status, data, static_cast<int64_t>(size));
 }

cpp/src/arrow/util/fuzz_internal.cc

@@ -34,7 +34,7 @@ MemoryPool* fuzzing_memory_pool() {
   return pool.get();
 }
 
-void LogFuzzStatus(const Status& st, const uint8_t* data, int64_t size) {
+int LogFuzzStatus(const FuzzStatus& st, const uint8_t* data, int64_t size) {
   static const int kVerbosity = []() {
     auto maybe_env_value =
         GetEnvVarInteger("ARROW_FUZZING_VERBOSITY", /*min_value=*/0, /*max_value=*/1);
@@ -48,13 +48,23 @@ void LogFuzzStatus(const Status& st, const uint8_t* data, int64_t size) {
     return 0;
   }();
 
-  if (!st.ok() && kVerbosity >= 1) {
-    ARROW_LOG(WARNING) << "Fuzzing input with size=" << size
-                       << " failed: " << st.ToString();
-  } else if (st.IsOutOfMemory()) {
+  const auto& reason = FuzzReason(st);
+  bool skip_input = std::holds_alternative<SkipFuzzInput>(st);
+  if (skip_input) {
+    if (kVerbosity >= 1) {
+      ARROW_LOG(WARNING) << "Skipping input with size=" << size << ": "
+                         << reason.ToString();
+    }
+  } else if (reason.IsOutOfMemory()) {
     ARROW_LOG(WARNING) << "Fuzzing input with size=" << size
-                       << " hit allocation failure: " << st.ToString();
+                       << " hit allocation failure: " << reason.ToString();
+  } else {
+    if (!reason.ok() && kVerbosity >= 1) {
+      ARROW_LOG(WARNING) << "Fuzzing input with size=" << size
+                         << " failed: " << reason.ToString();
+    }
   }
+  return skip_input ? -1 : 0;
 }
 
 }  // namespace arrow::internal

cpp/src/arrow/util/fuzz_internal.h

@@ -18,12 +18,33 @@
 #pragma once
 
 #include <cstdint>
+#include <variant>
 
+#include "arrow/status.h"
 #include "arrow/type_fwd.h"
 #include "arrow/util/macros.h"
 
 namespace arrow::internal {
 
+// A tag used to signal the fuzzing engine that this input should not be saved
+// into the corpus.
+struct SkipFuzzInput {
+  Status reason;
+};
+
+// The Status alternative holds a regular fuzzing success or failure,
+// while the SkipFuzzInput alternative holds a setup failure (e.g. invalid
+// fuzzing parameters encoded in the payload).
+using FuzzStatus = std::variant<Status, SkipFuzzInput>;
+
+inline const Status& FuzzReason(const FuzzStatus& st) {
+  struct Visitor {
+    const Status& operator()(const SkipFuzzInput& v) { return v.reason; }
+    const Status& operator()(const Status& v) { return v; }
+  };
+  return std::visit(Visitor{}, st);
+}
+
 // The default rss_limit_mb on OSS-Fuzz is 2560 MB and we want to fail allocations
 // before that limit is reached, otherwise the fuzz target gets killed (GH-48105).
 constexpr int64_t kFuzzingMemoryLimit = 2200LL * 1000 * 1000;
@@ -32,6 +53,12 @@ constexpr int64_t kFuzzingMemoryLimit = 2200LL * 1000 * 1000;
 ARROW_EXPORT MemoryPool* fuzzing_memory_pool();
 
 /// Optionally log the outcome of fuzzing an input
-ARROW_EXPORT void LogFuzzStatus(const Status&, const uint8_t* data, int64_t size);
+///
+/// Returns the integer code to return from LLVMFuzzerTestOneInput.
+ARROW_EXPORT int LogFuzzStatus(const FuzzStatus&, const uint8_t* data, int64_t size);
+
+inline int LogFuzzStatus(const Status& status, const uint8_t* data, int64_t size) {
+  return LogFuzzStatus(FuzzStatus{status}, data, size);
+}
 
 }  // namespace arrow::internal

cpp/src/parquet/arrow/encoding_fuzz.cc

@@ -22,6 +22,5 @@
 extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
   auto status =
       parquet::fuzzing::internal::FuzzEncoding(data, static_cast<int64_t>(size));
-  arrow::internal::LogFuzzStatus(status, data, static_cast<int64_t>(size));
-  return 0;
+  return arrow::internal::LogFuzzStatus(status, data, static_cast<int64_t>(size));
 }

cpp/src/parquet/arrow/fuzz.cc

@@ -21,6 +21,5 @@
 
 extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
   auto status = parquet::fuzzing::internal::FuzzReader(data, static_cast<int64_t>(size));
-  arrow::internal::LogFuzzStatus(status, data, static_cast<int64_t>(size));
-  return 0;
+  return arrow::internal::LogFuzzStatus(status, data, static_cast<int64_t>(size));
 }

cpp/src/parquet/arrow/fuzz_encoding_internal.cc

@@ -23,8 +23,10 @@
 #include <functional>
 #include <limits>
 #include <new>
+#include <optional>
 #include <sstream>
 #include <string_view>
+#include <typeinfo>
 
 #include "arrow/array.h"
 #include "arrow/array/builder_binary.h"
@@ -52,6 +54,7 @@ using ::arrow::MemoryPool;
 using ::arrow::Result;
 using ::arrow::Status;
 using ::arrow::TypedBufferBuilder;
+using ::arrow::internal::FuzzStatus;
 using ::parquet::arrow::FileReader;
 
 ColumnDescriptor MakeColumnDescriptor(Type::type type, int type_length) {
@@ -140,6 +143,10 @@ ::arrow::Result<FuzzEncodingHeader::ParseResult> FuzzEncodingHeader::Parse(
                             static_cast<Encoding::type>(ph.roundtrip_encoding_id),
                             static_cast<Type::type>(ph.type_id), ph.type_length,
                             ph.num_values);
+  if (!IsEncodingSupported(header.type, header.source_encoding) ||
+      !IsEncodingSupported(header.type, header.roundtrip_encoding)) {
+    return invalid_payload();
+  }
   if ((header.type == Type::FIXED_LEN_BYTE_ARRAY) ? (header.type_length <= 0)
                                                   : (header.type_length != -1)) {
     return invalid_payload();
@@ -492,13 +499,17 @@ struct TypedFuzzEncoding {
 
 }  // namespace
 
-Status FuzzEncoding(const uint8_t* data, int64_t size) {
+FuzzStatus FuzzEncoding(const uint8_t* data, int64_t size) {
   constexpr auto kInt32Max = std::numeric_limits<int32_t>::max();
 
-  ARROW_ASSIGN_OR_RAISE(const auto parse_result,
-                        FuzzEncodingHeader::Parse(std::span(data, size)));
-  const auto header = parse_result.first;
-  const auto encoded_data = parse_result.second;
+  auto maybe_parse_result = FuzzEncodingHeader::Parse(std::span(data, size));
+  if (!maybe_parse_result.ok()) {
+    // If the fuzz encoding header is invalid, we won't save this input
+    // in the corpus, because it didn't exercise anything interesting.
+    return ::arrow::internal::SkipFuzzInput{maybe_parse_result.status()};
+  }
+  const auto header = maybe_parse_result->first;
+  const auto encoded_data = maybe_parse_result->second;
   if (encoded_data.size() > static_cast<size_t>(kInt32Max)) {
     // Unlikely but who knows?
     return Status::Invalid("Fuzz payload too large");

cpp/src/parquet/arrow/fuzz_encoding_internal.h

@@ -27,6 +27,7 @@
 #include "arrow/result.h"
 #include "arrow/status.h"
 #include "arrow/type_fwd.h"
+#include "arrow/util/fuzz_internal.h"
 #include "arrow/util/macros.h"
 #include "parquet/platform.h"
 #include "parquet/types.h"
@@ -76,7 +77,8 @@ struct FuzzEncodingHeader {
 };
 
 /// Fuzz a payload encoded as explained in FuzzEncodingHeader
-PARQUET_EXPORT ::arrow::Status FuzzEncoding(const uint8_t* data, int64_t size);
+PARQUET_EXPORT ::arrow::internal::FuzzStatus FuzzEncoding(const uint8_t* data,
+                                                          int64_t size);
 
 PARQUET_EXPORT ColumnDescriptor MakeColumnDescriptor(Type::type type,
                                                      int type_length = -1);

cpp/src/parquet/decoder.cc

@@ -2557,27 +2557,50 @@ std::unique_ptr<Decoder> MakeDictDecoder(Type::type type_num,
 
 }  // namespace detail
 
-std::vector<Encoding::type> SupportedEncodings(Type::type physical_type) {
+namespace {
+
+// Discriminant allows definition of distinct functions for different initializers.
+template <auto Discriminant, typename T>
+const std::vector<T>& ConstVectorRef(std::initializer_list<T> init) {
+  static const auto vec = std::vector<T>(std::move(init));
+  return vec;
+}
+
+const std::vector<Encoding::type>& SupportedEncodingsRef(Type::type physical_type) {
   switch (physical_type) {
     case Type::BOOLEAN:
-      return {Encoding::PLAIN, Encoding::RLE};
+      return ConstVectorRef<Type::BOOLEAN>({Encoding::PLAIN, Encoding::RLE});
     case Type::INT32:
     case Type::INT64:
-      return {Encoding::PLAIN, Encoding::DELTA_BINARY_PACKED,
-              Encoding::BYTE_STREAM_SPLIT};
+      return ConstVectorRef<Type::INT32>(
+          {Encoding::PLAIN, Encoding::DELTA_BINARY_PACKED, Encoding::BYTE_STREAM_SPLIT});
     case Type::INT96:
-      return {Encoding::PLAIN};
+      return ConstVectorRef<Type::INT96>({Encoding::PLAIN});
     case Type::FLOAT:
     case Type::DOUBLE:
-      return {Encoding::PLAIN, Encoding::BYTE_STREAM_SPLIT};
+      return ConstVectorRef<Type::FLOAT>({Encoding::PLAIN, Encoding::BYTE_STREAM_SPLIT});
     case Type::FIXED_LEN_BYTE_ARRAY:
-      return {Encoding::PLAIN, Encoding::BYTE_STREAM_SPLIT, Encoding::DELTA_BYTE_ARRAY};
+      return ConstVectorRef<Type::FIXED_LEN_BYTE_ARRAY>(
+          {Encoding::PLAIN, Encoding::BYTE_STREAM_SPLIT, Encoding::DELTA_BYTE_ARRAY});
     case Type::BYTE_ARRAY:
-      return {Encoding::PLAIN, Encoding::DELTA_LENGTH_BYTE_ARRAY,
-              Encoding::DELTA_BYTE_ARRAY};
+      return ConstVectorRef<Type::BYTE_ARRAY>({Encoding::PLAIN,
+                                               Encoding::DELTA_LENGTH_BYTE_ARRAY,
+                                               Encoding::DELTA_BYTE_ARRAY});
     default:
       throw ParquetException("Invalid physical type");
   }
 }
 
+}  // namespace
+
+const std::vector<Encoding::type>& SupportedEncodings(Type::type physical_type) {
+  return SupportedEncodingsRef(physical_type);
+}
+
+bool IsEncodingSupported(Type::type physical_type, Encoding::type encoding) {
+  const auto& supported_encodings = SupportedEncodingsRef(physical_type);
+  return std::find(supported_encodings.begin(), supported_encodings.end(), encoding) !=
+         supported_encodings.end();
+}
+
 }  // namespace parquet

cpp/src/parquet/encoding.h

@@ -482,6 +482,12 @@ std::unique_ptr<typename EncodingTraits<DType>::Decoder> MakeTypedDecoder(
 ///
 /// Only non-dictionary encodings are returned.
 PARQUET_EXPORT
-std::vector<Encoding::type> SupportedEncodings(Type::type physical_type);
+const std::vector<Encoding::type>& SupportedEncodings(Type::type physical_type);
+
+/// Return whether the encoding is supported for the given physical type
+///
+/// Dictionary encodings return false.
+PARQUET_EXPORT
+bool IsEncodingSupported(Type::type physical_type, Encoding::type encoding);
 
 }  // namespace parquet

cpp/src/parquet/encoding_test.cc

@@ -77,8 +77,10 @@ void TestSupportedEncodingsConsistentWith(
       if (std::find(supported_encodings.begin(), supported_encodings.end(), encoding) !=
           supported_encodings.end()) {
         ASSERT_NO_THROW(func(type, encoding, descr));
+        ASSERT_TRUE(IsEncodingSupported(type, encoding));
       } else {
         ASSERT_THROW(func(type, encoding, descr), ParquetException);
+        ASSERT_FALSE(IsEncodingSupported(type, encoding));
       }
     }
   }