fix(manifest): reject unsupported format versions

manifest.go

@@ -546,10 +546,10 @@ type hasFieldToIDMap interface {
 	setFieldIDToDecimalScaleMap(map[int]int)
 }
 
-// ManifestFile is the interface which covers both V1 and V2 manifest files.
+// ManifestFile is the interface for version 1, 2, and 3 manifest files.
 type ManifestFile interface {
 	// Version returns the version number of this manifest file.
-	// It should be 1 or 2.
+	// It must be 1, 2, or 3.
 	Version() int
 	// FilePath is the location URI of this manifest file.
 	FilePath() string
@@ -744,6 +744,9 @@ func NewManifestReader(file ManifestFile, in io.Reader) (*ManifestReader, error)
 			return nil, fmt.Errorf("manifest file's 'format-version' metadata is invalid: %w", err)
 		}
 	}
+	if err := validateManifestFormatVersion(formatVersion); err != nil {
+		return nil, err
+	}
 	// The manifest's own metadata is authoritative for its version. A v2/v3
 	// manifest list may reference older manifests so a table can be upgraded
 	// in place without rewriting history; only a manifest newer than the list
@@ -1015,6 +1018,9 @@ func ReadManifestList(in io.Reader) ([]ManifestFile, error) {
 
 			version = v
 		}
+		if err := validateManifestFormatVersion(version); err != nil {
+			return nil, err
+		}
 
 		if version == 1 {
 			return manifestFileV1Reader, nil
@@ -1033,6 +1039,14 @@ func ReadManifestList(in io.Reader) ([]ManifestFile, error) {
 	return decodeManifests[*manifestFile](rd, version)
 }
 
+func validateManifestFormatVersion(version int) error {
+	if version < 1 || version > 3 {
+		return fmt.Errorf("unsupported manifest format version: %d", version)
+	}
+
+	return nil
+}
+
 type writerImpl interface {
 	prepareEntry(*manifestEntry, int64) (ManifestEntry, error)
 }

manifest_test.go

@@ -1026,6 +1026,23 @@ func (m *ManifestTestSuite) TestReadManifestListMissingFormatVersion() {
 	m.Empty(files) // the file has no entries, just headers
 }
 
+func (m *ManifestTestSuite) TestReadManifestListRejectsUnsupportedFormatVersion() {
+	for _, version := range []int{-1, 0, 4} {
+		m.Run(strconv.Itoa(version), func() {
+			fileSchema, err := internal.NewManifestFileSchema(2)
+			m.Require().NoError(err)
+			var buf bytes.Buffer
+			writer, err := ocf.NewWriter(&buf, fileSchema,
+				ocf.WithMetadata(map[string][]byte{"format-version": []byte(strconv.Itoa(version))}))
+			m.Require().NoError(err)
+			m.Require().NoError(writer.Close())
+
+			_, err = ReadManifestList(&buf)
+			m.ErrorContains(err, "unsupported manifest format version")
+		})
+	}
+}
+
 // writeManifestNoFormatVersion writes a valid v1 manifest entry Avro file that
 // omits the "format-version" metadata key, simulating files produced by the Java
 // Iceberg library (format-version is optional for v1 per the Iceberg spec).
@@ -1079,6 +1096,34 @@ func (m *ManifestTestSuite) TestNewManifestReaderMissingFormatVersion() {
 	m.NoError(reader.Close())
 }
 
+func (m *ManifestTestSuite) TestNewManifestReaderRejectsUnsupportedFormatVersion() {
+	for _, version := range []int{-1, 0, 4} {
+		m.Run(strconv.Itoa(version), func() {
+			spec := NewPartitionSpec()
+			partitionSchema, err := partitionTypeToAvroSchema(spec.PartitionType(testSchema))
+			m.Require().NoError(err)
+			entrySchema, err := internal.NewManifestEntrySchema(partitionSchema, 1)
+			m.Require().NoError(err)
+			schemaJSON, err := json.Marshal(testSchema)
+			m.Require().NoError(err)
+			var manifest bytes.Buffer
+			writer, err := ocf.NewWriter(&manifest, entrySchema, ocf.WithMetadata(map[string][]byte{
+				"format-version":    []byte(strconv.Itoa(version)),
+				"schema":            schemaJSON,
+				"schema-id":         []byte(strconv.Itoa(testSchema.ID)),
+				"partition-spec":    []byte("[]"),
+				"partition-spec-id": []byte("0"),
+				"content":           []byte("data"),
+			}))
+			m.Require().NoError(err)
+			m.Require().NoError(writer.Close())
+
+			_, err = NewManifestReader(&manifestFile{version: version}, &manifest)
+			m.ErrorContains(err, "unsupported manifest format version")
+		})
+	}
+}
+
 func (m *ManifestTestSuite) TestV3DataManifestFirstRowIDInheritance() {
 	// Build a v3 data manifest with two entries that have null first_row_id.
 	partitionSpec := NewPartitionSpecID(1,