This is an automated email from the ASF dual-hosted git repository. xiangfu0 pushed a commit to branch xiangfu0/codex2/ratis-offline-publication in repository https://gitbox.apache.org/repos/asf/pinot.git
commit cc871105703176db823e8c5d0a24f09ba718c826 Author: Xiang Fu <[email protected]> AuthorDate: Tue Oct 20 02:08:36 2026 -0700 Publish and reconcile native immutable OFFLINE uploads --- .../NativeControllerSegmentResource.java | 211 +++++++++ .../coordination/NativeControllerTasks.java | 77 +++- .../coordination/NativeSegmentUploadStorage.java | 489 +++++++++++++++++++++ .../coordination/NativeTaskPlacementStatus.java | 10 +- .../coordination/RatisControllerRuntime.java | 1 + .../coordination/NativeControllerTasksTest.java | 46 +- .../NativeSegmentUploadStorageTest.java | 346 +++++++++++++++ pinot-coordination-ratis/CONTROLLER-API-PARITY.md | 12 +- pinot-coordination-ratis/MINION-TASK-PLAN.md | 3 +- pinot-coordination-ratis/README.md | 53 ++- ...og.java => OfflineContentPlacementCatalog.java} | 42 +- .../ratis/metadata/OfflineSegmentCatalog.java | 30 +- .../ratis/metadata/OfflineSegmentProtocol.java | 17 +- .../ratis/metadata/TaskCatalogTest.java | 80 ++++ .../pinot/tools/service/RatisStartersTest.java | 221 +++++----- 15 files changed, 1479 insertions(+), 159 deletions(-) diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerSegmentResource.java b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerSegmentResource.java new file mode 100644 index 00000000000..df5c9bfde7f --- /dev/null +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerSegmentResource.java @@ -0,0 +1,211 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.controller.coordination; + +import java.io.IOException; +import java.io.InputStream; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Semaphore; +import javax.ws.rs.Consumes; +import javax.ws.rs.DefaultValue; +import javax.ws.rs.POST; +import javax.ws.rs.Path; +import javax.ws.rs.Produces; +import javax.ws.rs.QueryParam; +import javax.ws.rs.WebApplicationException; +import javax.ws.rs.core.Context; +import javax.ws.rs.core.HttpHeaders; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.Response; +import org.apache.pinot.common.utils.DatabaseUtils; +import org.apache.pinot.common.utils.FileUploadDownloadClient.CustomHeaders; +import org.apache.pinot.controller.ControllerConf; +import org.apache.pinot.controller.api.access.AccessControlFactory; +import org.apache.pinot.controller.api.access.AccessControlUtils; +import org.apache.pinot.controller.api.access.AccessType; +import org.apache.pinot.controller.api.access.Authenticate; +import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ObjectVersion; +import org.apache.pinot.coordination.ratis.metadata.NativeReplicaPlacementPolicy; +import org.apache.pinot.core.auth.Actions; +import org.apache.pinot.core.auth.Authorize; +import org.apache.pinot.core.auth.TargetType; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.exception.DatabaseConflictException; +import org.apache.pinot.spi.utils.CommonConstants; +import org.apache.pinot.spi.utils.JsonUtils; +import org.apache.pinot.spi.utils.builder.TableNameBuilder; + + +/// Standard binary OFFLINE upload with explicit immutable publication semantics. Parses multipart only after +/// authorization and bounded admission; no legacy Helix upload resource or global multipart temp directory is used. +@Path("/v2/segments") +@Produces(MediaType.APPLICATION_JSON) +public final class NativeControllerSegmentResource { + private final NativeControllerMetadata _metadata; + private final NativeControllerTasks _tasks; + private final AccessControlFactory _access; + private final NativeSegmentUploadStorage _storage; + private final Semaphore _requests = new Semaphore(2); + + public NativeControllerSegmentResource(NativeControllerMetadata metadata, NativeControllerTasks tasks, + ControllerConf conf, AccessControlFactory access) throws IOException { + _metadata = metadata; + _tasks = tasks; + _access = access; + _storage = new NativeSegmentUploadStorage(conf); + } + + @POST + @Consumes(MediaType.MULTIPART_FORM_DATA) + @Authenticate(AccessType.CREATE) + @Authorize(targetType = TargetType.TABLE, paramName = "tableName", action = Actions.Table.UPLOAD_SEGMENT) + public Response upload(InputStream body, @QueryParam("tableName") String tableName, + @DefaultValue("OFFLINE") @QueryParam("tableType") String tableType, + @DefaultValue("false") @QueryParam("allowRefresh") boolean allowRefresh, + @DefaultValue("false") @QueryParam("enableParallelPushProtection") boolean parallelProtection, + @Context HttpHeaders headers) { + String table = tableName(tableName, tableType, headers); + authorize(table, headers); + String headerTable = headers.getHeaderString(CommonConstants.Controller.TABLE_NAME_HTTP_HEADER); + if (headerTable != null && !table.equals(tableName(headerTable, tableType, headers))) { + throw error(400, "Upload table header differs from tableName"); + } + if (allowRefresh) { + throw error(400, "Native uploads do not support refresh or replacement; use allowRefresh=false"); + } + validateHeaders(headers); + if (!_storage.enabled()) { + throw error(503, "Native shared upload storage is not configured"); + } + String length = headers.getHeaderString(HttpHeaders.CONTENT_LENGTH); + try { + if (length != null && (Long.parseLong(length) < 0 || Long.parseLong(length) > _storage.maxRequestBytes())) { + throw error(413, "Upload exceeds its byte limit"); + } + } catch (NumberFormatException e) { + throw error(400, "Invalid request length"); + } + if (!_requests.tryAcquire()) { + throw error(503, "Native upload capacity is exhausted"); + } + try { + var snapshot = _metadata.snapshot(); + var tableRecord = snapshot.tables().get(table); + if (tableRecord == null) { + throw error(404, "Table not found"); + } + var schemaRecord = snapshot.schemas().get(TableNameBuilder.extractRawTableName(table)); + if (schemaRecord == null) { + throw error(409, "Table schema is unavailable"); + } + TableConfig config = JsonUtils.jsonNodeToObject(tableRecord.value(), TableConfig.class); + NativeReplicaPlacementPolicy.validateTable(config); + validateUploadTable(config); + Schema schema = JsonUtils.jsonNodeToObject(schemaRecord.value(), Schema.class); + var history = snapshot.revision().historyId(); + var content = _storage.store(body, headers.getMediaType().getParameters().get("boundary"), history, + tableRecord.objectId(), config, schema); + // Immutable name/content CAS always supplies parallel push protection, regardless of the legacy flag. + authorize(table, headers); + // The replicated operation checks these exact versions and the current controller owner atomically. + var result = _tasks.publish(content, new ObjectVersion(history, tableRecord.objectId(), tableRecord.version()), + new ObjectVersion(history, schemaRecord.objectId(), schemaRecord.version())); + int status = switch (result.status()) { + case OK -> 200; + case INVALID_REQUEST -> 400; + case NOT_FOUND -> 404; + case CONFLICT, FENCED, HISTORY_MISMATCH -> 409; + default -> 503; + }; + String message = status == 200 ? "Successfully uploaded segment: " + content.segmentName() + " of table: " + table + : result.status().name(); + return Response.status(status).entity(JsonUtils.objectToString(Map.of("status", message, + "segmentName", content.segmentName(), "segmentObjectId", content.segmentObjectId(), + "published", status == 200))).header(HttpHeaders.CACHE_CONTROL, "no-store").build(); + } catch (WebApplicationException e) { + throw e; + } catch (IllegalArgumentException e) { + throw error(400, "Invalid or unsupported segment upload"); + } catch (IOException e) { + return error(503, "Native upload publication is unavailable; retry the identical archive").getResponse(); + } finally { + _requests.release(); + } + } + + private static String tableName(String name, String type, HttpHeaders headers) { + try { + if (name == null || name.isBlank() || !"OFFLINE".equalsIgnoreCase(type)) { + throw error(400, "An explicit OFFLINE tableName is required"); + } + String translated = DatabaseUtils.translateTableName(name, headers); + String raw = TableNameBuilder.extractRawTableName(translated); + if (TableNameBuilder.getTableTypeFromTableName(translated) == TableType.REALTIME + || raw == null || raw.isBlank() || TableNameBuilder.getTableTypeFromTableName(raw) != null) { + throw error(400, "Invalid OFFLINE table name"); + } + return TableNameBuilder.OFFLINE.tableNameWithType(translated); + } catch (IllegalArgumentException | DatabaseConflictException e) { + throw error(400, "Invalid table name"); + } + } + + private void authorize(String table, HttpHeaders headers) { + var access = _access.create(); + String raw = TableNameBuilder.extractRawTableName(table); + AccessControlUtils.validatePermission(raw, AccessType.CREATE, headers, "/v2/segments", access); + if (!access.hasAccess(headers, TargetType.TABLE, raw, Actions.Table.UPLOAD_SEGMENT)) { + throw error(403, "Upload permission denied"); + } + } + + private static void validateHeaders(HttpHeaders headers) { + String uploadType = headers.getHeaderString(CustomHeaders.UPLOAD_TYPE); + if (uploadType != null && !"SEGMENT".equalsIgnoreCase(uploadType)) { + throw error(400, "Native upload supports binary SEGMENT requests only"); + } + for (String header : List.of(CustomHeaders.REFRESH_ONLY, CustomHeaders.DOWNLOAD_URI, + CustomHeaders.COPY_SEGMENT_TO_DEEP_STORE, CustomHeaders.SEGMENT_ZK_METADATA_CUSTOM_MAP_MODIFIER, + CustomHeaders.CRYPTER, CommonConstants.Controller.INGESTION_DESCRIPTOR, HttpHeaders.IF_MATCH, + HttpHeaders.IF_NONE_MATCH)) { + if (headers.getHeaderString(header) != null) { + throw error(400, "Unsupported native upload option"); + } + } + } + + @SuppressWarnings("deprecation") // Explicitly reject persisted legacy refresh controls as well as modern controls. + private static void validateUploadTable(TableConfig config) { + var batch = config.getIngestionConfig() == null ? null : config.getIngestionConfig().getBatchIngestionConfig(); + if (config.getQuotaConfig() != null || config.getValidationConfig().getCrypterClassName() != null + || "REFRESH".equalsIgnoreCase(config.getValidationConfig().getSegmentPushType()) + || batch != null && ("REFRESH".equalsIgnoreCase(batch.getSegmentIngestionType()) + || batch.getConsistentDataPush())) { + throw error(501, "Native upload does not yet enforce quotas, encryption, refresh or consistent batch push"); + } + } + + private static WebApplicationException error(int status, String message) { + return new WebApplicationException(message, Response.status(status).type(MediaType.APPLICATION_JSON_TYPE) + .header(HttpHeaders.CACHE_CONTROL, "no-store").entity(Map.of("error", message)).build()); + } +} diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerTasks.java b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerTasks.java index 72cbed96f01..3e44f5dbdb2 100644 --- a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerTasks.java +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeControllerTasks.java @@ -22,7 +22,6 @@ import java.io.IOException; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; -import java.util.ArrayList; import java.util.Comparator; import java.util.EnumSet; import java.util.List; @@ -50,8 +49,10 @@ import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Command; import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.FeedScope; import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.MembershipCommand; import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.MembershipQuery; +import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.ObjectVersion; import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.RecordKind; import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Result; +import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Revision; import org.apache.pinot.coordination.ratis.metadata.MetadataProtocol.Status; import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol; import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Content; @@ -67,15 +68,16 @@ import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.TaskState; import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.TaskStatus; import org.apache.pinot.coordination.ratis.transport.MembershipAccess; import org.apache.pinot.coordination.ratis.transport.MetadataAccess; +import org.apache.pinot.spi.utils.builder.TableNameBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import static com.google.common.base.Preconditions.checkArgument; -/// Bounded native task scheduling using committed membership rounds, never local wall-clock authority. A controller +/// Bounded native task scheduling and OFFLINE content reconciliation using committed membership rounds. A controller /// claims only unowned or expired table shards. Commands and heartbeats retain their identity across unknown replies. -/// Methods serialize the controller request stream; immutable schedule identities survive process restart. +/// Methods serialize the controller request stream; immutable schedule and content identities survive process restart. public final class NativeControllerTasks implements AutoCloseable { private static final Logger LOGGER = LoggerFactory.getLogger(NativeControllerTasks.class); private static final FeedScope SCOPE = new FeedScope(EnumSet.of(RecordKind.TABLE_CONFIG, RecordKind.SCHEMA, @@ -166,28 +168,18 @@ public final class NativeControllerTasks implements AutoCloseable { Map<UUID, CatalogRecord> records = image.records().stream() .collect(Collectors.toMap(CatalogRecord::objectId, record -> record)); NativeTaskPlacementStatus placement = new NativeTaskPlacementStatus(image); - List<Candidate> candidates = new ArrayList<>(); - for (CatalogRecord record : image.records()) { - if (record.kind() == RecordKind.TASK_SCHEDULE || record.kind() == RecordKind.TASK) { - candidates.add(new Candidate(record, null)); - } else if (record.kind() == RecordKind.TASK_PUBLICATION) { - for (Content output : TaskProtocol.decodePublication(record.value()).outputs()) { - candidates.add(new Candidate(record, output)); - } - } - } - candidates.sort(Comparator.comparing(candidate -> candidate.record().objectId() + "/" - + (candidate.output() == null ? "" : candidate.output().segmentObjectId()))); + List<CatalogRecord> candidates = image.records().stream() + .filter(record -> record.kind() == RecordKind.TASK_SCHEDULE || record.kind() == RecordKind.TASK + || record.kind() == RecordKind.OFFLINE_SEGMENT_CONTENT) + .sorted(Comparator.comparing(record -> record.objectId().toString())).toList(); int start = candidates.isEmpty() ? 0 : _scanCursor % candidates.size(); for (int offset = 0; offset < candidates.size(); offset++) { int position = (start + offset) % candidates.size(); _scanCursor = (position + 1) % candidates.size(); - Candidate candidate = candidates.get(position); - CatalogRecord record = candidate.record(); - if (candidate.output() != null) { - var publication = TaskProtocol.decodePublication(record.value()); - Content output = candidate.output(); - NativeTaskPlacementStatus.Output status = placement.output(publication.taskId(), output); + CatalogRecord record = candidates.get(position); + if (record.kind() == RecordKind.OFFLINE_SEGMENT_CONTENT) { + Content output = OfflineSegmentProtocol.decodeContent(record.value()); + NativeTaskPlacementStatus.Output status = placement.output(null, output); if (!status.needsReconciliation()) { continue; } @@ -195,7 +187,7 @@ public final class NativeControllerTasks implements AutoCloseable { if (owner != null) { Result reconciled = command(sequence -> new MetadataProtocol.OfflineSegmentCommand( _history, _clientId, 1, "tasks", sequence, - new OfflineSegmentProtocol.ReconcileTaskOutput(owner, publication.taskId(), output.segmentObjectId()))); + new OfflineSegmentProtocol.ReconcileContent(owner, output.segmentObjectId()))); require(reconciled.status() == Status.OK, "Native output reconciliation did not commit: " + reconciled.status()); return; @@ -252,7 +244,46 @@ public final class NativeControllerTasks implements AutoCloseable { } } - private record Candidate(CatalogRecord record, @Nullable Content output) { + /// Publishes a validated immutable archive using the same fenced owner and durable request stream as reconciliation. + /// Archive validation and file I/O must finish outside this monitor; replicated apply checks both captured versions. + public synchronized Result publish(Content content, ObjectVersion expectedTable, ObjectVersion expectedSchema) + throws IOException { + ensureOpen(); + checkArgument(content != null && expectedTable != null && expectedSchema != null, + "Content and validated table/schema versions are required"); + resolvePending(); + heartbeat(); + CatalogImage image = image(); + Revision revision = new Revision(_history, image.cursor().index()); + CatalogRecord table = record(image, RecordKind.TABLE_CONFIG, content.tableName()); + CatalogRecord schema = record(image, RecordKind.SCHEMA, + TableNameBuilder.extractRawTableName(content.tableName())); + if (!matches(table, expectedTable) || !matches(schema, expectedSchema) + || !content.tableObjectId().equals(expectedTable.objectId())) { + return new Result(Status.CONFLICT, revision, "Upload validation dependencies changed", null, null, null); + } + CatalogRecord existing = record(image, RecordKind.OFFLINE_SEGMENT_CONTENT, + OfflineSegmentProtocol.contentName(content.tableObjectId(), content.segmentName())); + if (existing != null) { + boolean identical = !existing.deleted() && existing.objectId().equals(content.segmentObjectId()) + && content.equals(OfflineSegmentProtocol.decodeContent(existing.value())); + // Any controller can resolve an existing immutable publication through a committed read barrier. This + // does not create new content or acquire another controller's live shard ownership. + return new Result(identical ? Status.OK : Status.CONFLICT, revision, + identical ? "Immutable content already published" : "Segment name identifies different immutable content", + null, null, existing); + } + OwnerToken owner = owner(image, content.tableName()); + if (owner == null) { + throw new IOException("Segment table is owned by another controller; submit to that controller"); + } + return command(sequence -> new MetadataProtocol.OfflineSegmentCommand(_history, _clientId, 1, "tasks", sequence, + new OfflineSegmentProtocol.PublishValidatedContent(owner, expectedTable, expectedSchema, content))); + } + + private boolean matches(@Nullable CatalogRecord record, ObjectVersion version) { + return record != null && !record.deleted() && _history.equals(version.historyId()) + && record.objectId().equals(version.objectId()) && record.version() == version.version(); } public synchronized Result schedule(String tableName, UUID scheduleId, ScheduleRequest request) throws IOException { diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeSegmentUploadStorage.java b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeSegmentUploadStorage.java new file mode 100644 index 00000000000..be3d60ad861 --- /dev/null +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeSegmentUploadStorage.java @@ -0,0 +1,489 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.controller.coordination; + +import java.io.FilterInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.PushbackInputStream; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.charset.StandardCharsets; +import java.nio.file.FileAlreadyExistsException; +import java.nio.file.Files; +import java.nio.file.LinkOption; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.security.DigestInputStream; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.text.ParseException; +import java.util.HashSet; +import java.util.HexFormat; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import javax.annotation.Nullable; +import org.apache.commons.compress.archivers.tar.TarArchiveEntry; +import org.apache.commons.compress.archivers.tar.TarArchiveInputStream; +import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream; +import org.apache.commons.configuration2.PropertiesConfiguration; +import org.apache.commons.configuration2.ex.ConfigurationException; +import org.apache.commons.configuration2.ex.ConfigurationRuntimeException; +import org.apache.commons.io.FileUtils; +import org.apache.pinot.common.coordination.NativeMetadataConfig; +import org.apache.pinot.common.utils.LLCSegmentName; +import org.apache.pinot.controller.ControllerConf; +import org.apache.pinot.controller.api.upload.SegmentValidationUtils; +import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Content; +import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.LoaderDescriptor; +import org.apache.pinot.segment.local.utils.CrcUtils; +import org.apache.pinot.segment.spi.V1Constants; +import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.data.FieldSpec; +import org.apache.pinot.spi.data.Schema; +import org.glassfish.jersey.media.multipart.FormDataContentDisposition; +import org.jvnet.mimepull.MIMEConfig; +import org.jvnet.mimepull.MIMEMessage; +import org.jvnet.mimepull.MIMEParsingException; +import org.jvnet.mimepull.MIMEPart; + +import static com.google.common.base.Preconditions.checkArgument; + + +/// Validates bounded uploads and durably links immutable archives into an explicitly configured shared filesystem. +/// Each invocation owns its staging directory. Final objects are never overwritten or deleted on an uncertain commit. +final class NativeSegmentUploadStorage { + static final String PREFIX = NativeMetadataConfig.PREFIX + "controller.upload."; + private static final int MULTIPART_OVERHEAD_BYTES = 64 * 1024; + @Nullable + private final Path _root; + private final long _maxArchiveBytes; + private final long _maxExpandedBytes; + private final int _maxEntries; + + NativeSegmentUploadStorage(ControllerConf conf) throws IOException { + _maxArchiveBytes = conf.getProperty(PREFIX + "maxArchiveBytes", 128L * 1024 * 1024); + _maxExpandedBytes = conf.getProperty(PREFIX + "maxExpandedBytes", 512L * 1024 * 1024); + _maxEntries = conf.getProperty(PREFIX + "maxEntries", 4096); + checkArgument(_maxArchiveBytes > 0 && _maxArchiveBytes <= 512L * 1024 * 1024 + && _maxExpandedBytes > 0 && _maxExpandedBytes <= 2L * 1024 * 1024 * 1024 + && _maxEntries > 0 && _maxEntries <= 16384, "Native upload limits exceed supported bounds"); + String root = conf.getProperty(PREFIX + "storageRoot"); + if (root == null || root.isBlank()) { + _root = null; + } else { + Path path = Path.of(root); + checkArgument(path.isAbsolute() && path.equals(path.normalize()), "Upload root must be absolute and normalized"); + _root = path.toRealPath(); + verifyDirectories(_root); + } + } + + boolean enabled() { + return _root != null; + } + + long maxRequestBytes() { + return _maxArchiveBytes + MULTIPART_OVERHEAD_BYTES; + } + + Content store(InputStream entity, String boundary, UUID history, UUID tableId, TableConfig table, Schema schema) + throws IOException { + checkArgument(_root != null, "Native upload storage is not configured"); + checkArgument(boundary != null && !boundary.isBlank() && boundary.length() <= 200 + && boundary.chars().allMatch(c -> c >= 32 && c < 127), "Invalid multipart boundary"); + verifyDirectories(_root); + Path temporary = Files.createTempDirectory(_root, ".upload-"); + try { + Path archive = temporary.resolve("archive.tar.gz"); + String digest = receive(entity, boundary, temporary, archive); + Path extracted = Files.createDirectory(temporary.resolve("extracted")); + Path segment = extract(archive, extracted); + SegmentMetadataImpl metadata; + try { + metadata = new SegmentMetadataImpl(segment.toFile()); + validate(metadata, segment, table, schema); + } catch (IOException e) { + throw new IllegalArgumentException("Invalid Pinot segment metadata", e); + } catch (ConfigurationException e) { + throw new IllegalArgumentException("Invalid Pinot segment metadata", e); + } + Path directory = _root.resolve(history.toString()).resolve(tableId.toString()); + createDirectories(directory); + Path destination = directory.resolve(digest + ".tar.gz"); + try { + // Linking a fully fsynced inode is atomic and fails if another controller already published this digest. + Files.createLink(destination, archive); + } catch (FileAlreadyExistsException e) { + checkArgument(Files.isRegularFile(destination, LinkOption.NOFOLLOW_LINKS) + && Files.size(destination) == Files.size(archive) && digest.equals(digest(destination)), + "Existing immutable upload artifact does not match its digest"); + } + for (Path parent = directory; parent.startsWith(_root); parent = parent.getParent()) { + try (FileChannel channel = FileChannel.open(parent, StandardOpenOption.READ)) { + channel.force(true); + } + } + UUID objectId = UUID.nameUUIDFromBytes((history + "\0NATIVE_UPLOAD\0" + tableId + "\0" + + metadata.getName() + "\0" + digest).getBytes(StandardCharsets.UTF_8)); + // Data CRC predates v3 index packing. Use the verified full-segment CRC, not an unverified data-CRC claim. + return new Content(table.getTableName(), tableId, metadata.getName(), objectId, digest, + destination.toUri().toString(), new LoaderDescriptor(metadata.getCrc(), -1, false, null, null, + metadata.getIndexCreationTime(), -1, Map.of())); + } finally { + FileUtils.deleteDirectory(temporary.toFile()); + } + } + + private String receive(InputStream entity, String boundary, Path temporary, Path archive) throws IOException { + MIMEConfig config = new MIMEConfig(); + config.setMemoryThreshold(8192); + config.setDir(temporary.toString()); + MessageDigest digest = sha256(); + try (LimitedInputStream bounded = new LimitedInputStream(entity, MULTIPART_OVERHEAD_BYTES); + MIMEMessage message = new MIMEMessage(new FilterInputStream(bounded) { + @Override + public void close() { + // MIMEPull closes on END_MESSAGE. This method owns the entity until its bounded epilogue is consumed. + } + }, boundary, config)) { + MIMEPart part = message.getPart(0); + List<String> dispositions = part.getHeader("Content-Disposition"); + checkArgument(dispositions != null && dispositions.size() == 1 + && dispositions.getFirst().length() <= 8192, "Upload requires one file part"); + FormDataContentDisposition disposition = new FormDataContentDisposition(dispositions.getFirst()); + checkArgument("form-data".equalsIgnoreCase(disposition.getType()) && disposition.getName() != null + && disposition.getFileName() != null, "Upload requires one file part"); + String encoding = part.getContentTransferEncoding(); + checkArgument(encoding == null || "binary".equalsIgnoreCase(encoding) || "8bit".equalsIgnoreCase(encoding) + || "7bit".equalsIgnoreCase(encoding), "Encoded multipart parts are unsupported"); + bounded.limit(maxRequestBytes()); + long archiveBytes; + try (InputStream input = new LimitedInputStream(part.readOnce(), _maxArchiveBytes); + FileChannel output = FileChannel.open(archive, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE)) { + byte[] buffer = new byte[8192]; + for (int count; (count = input.read(buffer)) != -1;) { + digest.update(buffer, 0, count); + ByteBuffer bytes = ByteBuffer.wrap(buffer, 0, count); + while (bytes.hasRemaining()) { + output.write(bytes); + } + } + output.force(true); + archiveBytes = output.size(); + } + bounded.limit(archiveBytes + MULTIPART_OVERHEAD_BYTES); + checkArgument(message.getAttachments().size() == 1, "Upload requires exactly one file part"); + byte[] buffer = new byte[8192]; + while (bounded.read(buffer) != -1) { + // Bound any MIME epilogue as well as the file. + } + } catch (MIMEParsingException e) { + throw new IllegalArgumentException("Invalid or oversized multipart upload", e); + } catch (ParseException e) { + throw new IllegalArgumentException("Invalid multipart file disposition", e); + } + return HexFormat.of().formatHex(digest.digest()); + } + + private Path extract(Path archive, Path extracted) throws IOException { + Set<Path> entries = new HashSet<>(); + String root = null; + int count = 0; + long bytes = 0; + // Also bound decompressor output consumed internally while parsing tar headers, including PAX headers. + long tarLimit = _maxExpandedBytes + (long) _maxEntries * 1024 + MULTIPART_OVERHEAD_BYTES; + try (InputStream file = Files.newInputStream(archive); + InputStream gzip = GzipCompressorInputStream.builder().setInputStream(file) + .setDecompressConcatenated(true).get(); + InputStream bounded = new LimitedInputStream(gzip, tarLimit); + TarArchiveInputStream tar = new BoundedTarInputStream(bounded, _maxEntries)) { + for (TarArchiveEntry entry; (entry = tar.getNextEntry()) != null;) { + checkArgument(++count <= _maxEntries && (entry.isDirectory() || entry.isFile()) && !entry.isSparse() + && !entry.isSymbolicLink() && !entry.isLink() && tar.canReadEntryData(entry), + "Unsupported archive entry or excessive entry count"); + String name = entry.getName(); + checkArgument(name != null && !name.isBlank() && name.length() <= 4096 && !name.contains("\\") + && !name.startsWith("/") && name.chars().noneMatch(c -> c < 32), "Invalid archive entry path"); + Path relative = Path.of(name); + checkArgument(!relative.isAbsolute() && relative.normalize().equals(relative) + && !relative.startsWith("..") && relative.getNameCount() <= 16, "Archive path escapes its segment"); + if (root == null) { + root = relative.getName(0).toString(); + } + checkArgument(relative.getName(0).toString().equals(root) && entries.add(relative), + "Archive requires one segment directory without duplicate entries"); + Path destination = extracted.resolve(relative); + checkArgument(destination.startsWith(extracted) && (relative.getNameCount() > 1 || entry.isDirectory()), + "Archive requires a segment directory"); + if (entry.isDirectory()) { + Files.createDirectories(destination); + } else { + long size = entry.getSize(); + checkArgument(size >= 0 && size <= _maxExpandedBytes - bytes, "Expanded archive exceeds its byte limit"); + String fileName = relative.getFileName().toString(); + if (fileName.equals(V1Constants.MetadataKeys.METADATA_FILE_NAME) + || fileName.equals(V1Constants.INDEX_MAP_FILE_NAME)) { + checkArgument(size <= 2 * 1024 * 1024, "Segment metadata exceeds its byte limit"); + } + if (fileName.equals(V1Constants.SEGMENT_CREATION_META)) { + checkArgument(size == 16 || size == 24, "Invalid segment creation metadata"); + } + bytes += size; + Files.createDirectories(destination.getParent()); + try (var output = Files.newOutputStream(destination, StandardOpenOption.CREATE_NEW)) { + tar.transferTo(output); + } + checkArgument(Files.size(destination) == size, "Truncated archive entry"); + if (fileName.equals(V1Constants.MetadataKeys.METADATA_FILE_NAME) + || fileName.equals(V1Constants.INDEX_MAP_FILE_NAME)) { + validateProperties(destination); + } + } + } + // Reach the gzip trailer so its own checksum is verified; only tar padding may remain. + byte[] padding = new byte[8192]; + for (int size; (size = bounded.read(padding)) != -1;) { + for (int i = 0; i < size; i++) { + checkArgument(padding[i] == 0, "Archive contains trailing non-padding content"); + } + } + } catch (IOException e) { + throw new IllegalArgumentException("Invalid or oversized segment archive", e); + } + checkArgument(root != null, "Empty segment archive"); + return extracted.resolve(root); + } + + private static void validateProperties(Path file) throws IOException { + // Decode escapes and continuations exactly as the default segment reader, without loading a configuration: + // includes and interpolation must be rejected before SegmentMetadataImpl can evaluate either mechanism. + try (var reader = new PropertiesConfiguration.PropertiesReader(new InputStreamReader(Files.newInputStream(file)))) { + while (reader.nextProperty()) { + String key = reader.getPropertyName(); + String value = reader.getPropertyValue(); + checkArgument(!key.equalsIgnoreCase("include") && !key.equalsIgnoreCase("includeOptional") + && !key.equalsIgnoreCase(PropertiesConfiguration.getInclude()) + && !key.equalsIgnoreCase(PropertiesConfiguration.getIncludeOptional()) + && !key.contains("${") && !value.contains("${"), "Segment properties must be literal without includes"); + } + } catch (ConfigurationRuntimeException e) { + throw new IllegalArgumentException("Invalid segment properties", e); + } + } + + private static void validate(SegmentMetadataImpl metadata, Path segment, TableConfig table, Schema schema) + throws IOException { + String name = metadata.getName(); + checkArgument(name != null && !name.isBlank() && name.length() <= 1024 && !name.contains("/") + && !name.contains("\\") && !name.equals(".") && !name.equals("..") + && name.chars().noneMatch(c -> c < 32) && LLCSegmentName.of(name) == null, + "Invalid OFFLINE segment name"); + checkArgument(metadata.getTotalDocs() >= 0 && metadata.getCrc() >= 0 && metadata.getIndexCreationTime() >= -1, + "Segment requires complete creation metadata"); + checkArgument(metadata.getCrc() == CrcUtils.computeCrc(segment.toFile()), "Segment CRC does not match its bytes"); + Schema actual = metadata.getSchema(); + checkArgument(actual.getPhysicalColumnNames().equals(schema.getPhysicalColumnNames()), + "Segment columns differ from the destination schema"); + for (String column : schema.getPhysicalColumnNames()) { + FieldSpec expected = schema.getFieldSpecFor(column); + FieldSpec observed = actual.getFieldSpecFor(column); + checkArgument(expected.getDataType() == observed.getDataType() + && expected.isSingleValueField() == observed.isSingleValueField() + && expected.getFieldType() == observed.getFieldType(), "Segment column differs from destination schema"); + } + if (table.getIngestionConfig() == null || table.getIngestionConfig().isSegmentTimeValueCheck()) { + SegmentValidationUtils.validateTimeInterval(metadata, table); + } + } + + private void createDirectories(Path directory) throws IOException { + Path current = _root; + for (Path component : _root.relativize(directory)) { + current = current.resolve(component); + try { + Files.createDirectory(current); + } catch (FileAlreadyExistsException e) { + checkArgument(Files.isDirectory(current, LinkOption.NOFOLLOW_LINKS), "Upload path is not a directory"); + } + } + verifyDirectories(directory); + } + + private static void verifyDirectories(Path path) throws IOException { + for (Path current = path; current != null; current = current.getParent()) { + checkArgument(Files.isDirectory(current, LinkOption.NOFOLLOW_LINKS), + "Upload root must exist without symbolic links"); + } + } + + private String digest(Path file) throws IOException { + MessageDigest digest = sha256(); + try (InputStream input = new DigestInputStream( + new LimitedInputStream(Files.newInputStream(file, LinkOption.NOFOLLOW_LINKS), _maxArchiveBytes), digest)) { + byte[] buffer = new byte[8192]; + while (input.read(buffer) != -1) { + // Validate a competing writer's complete immutable inode before referring to it. + } + } + return HexFormat.of().formatHex(digest.digest()); + } + + private static MessageDigest sha256() { + try { + return MessageDigest.getInstance("SHA-256"); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException("SHA-256 is unavailable", e); + } + } + + /// Commons parses extension entries internally before returning a file. Check their headers and bounded PAX + /// framing first, preventing oversized allocations, recursive GNU header chains and sparse-map parsing. + private static final class BoundedTarInputStream extends TarArchiveInputStream { + private static final int EXTENSION_LIMIT = 64 * 1024; + private final PushbackInputStream _source; + private final int _headerLimit; + private int _headers; + private int _consecutiveExtensions; + private long _extensionBytes; + private boolean _sawEof; + + private BoundedTarInputStream(InputStream source, int maxEntries) { + this(new PushbackInputStream(source, EXTENSION_LIMIT), maxEntries); + } + + private BoundedTarInputStream(PushbackInputStream source, int maxEntries) { + // Do not let Commons silently skip unvalidated 10-KiB end padding. + super(source, 512); + _source = source; + _headerLimit = maxEntries * 5; + } + + @Override + protected byte[] readRecord() throws IOException { + byte[] record = super.readRecord(); + if (record == null || isEOFRecord(record)) { + _sawEof = true; + return record; + } + checkArgument(!_sawEof && ++_headers <= _headerLimit, "Archive contains excessive or trailing headers"); + TarArchiveEntry header = new TarArchiveEntry(record); + checkArgument(header.isCheckSumOK() && !header.isSparse() && !header.isSymbolicLink() && !header.isLink(), + "Invalid archive header or unsupported links/sparse content"); + boolean pax = header.isPaxHeader() || header.isGlobalPaxHeader(); + boolean extended = pax || header.isGNULongNameEntry() || header.isGNULongLinkEntry(); + if (!extended) { + _consecutiveExtensions = 0; + return record; + } + long size = header.getSize(); + checkArgument(size >= 0 && size <= EXTENSION_LIMIT - _extensionBytes + && (pax || size <= 16 * 1024) && ++_consecutiveExtensions <= 4, + "Archive extension metadata exceeds its bounds"); + _extensionBytes += size; + if (pax) { + byte[] metadata = _source.readNBytes((int) size); + checkArgument(metadata.length == size, "Truncated archive extension metadata"); + validatePax(metadata); + _source.unread(metadata); + } + return record; + } + + private static void validatePax(byte[] metadata) { + int offset = 0; + while (offset < metadata.length) { + int space = offset; + while (space < metadata.length && metadata[space] != ' ' && space - offset < 10) { + checkArgument(metadata[space] >= '0' && metadata[space] <= '9', "Invalid PAX record length"); + space++; + } + checkArgument(space > offset && space < metadata.length && metadata[space] == ' ', + "Invalid PAX record length"); + int length = Integer.parseInt(new String(metadata, offset, space - offset, StandardCharsets.US_ASCII)); + checkArgument(length > space - offset + 2 && length <= metadata.length - offset + && metadata[offset + length - 1] == '\n', "Invalid PAX record bounds"); + String value = new String(metadata, space + 1, offset + length - space - 2, StandardCharsets.UTF_8); + int equals = value.indexOf('='); + checkArgument(equals > 0 && equals <= 128, "Invalid PAX metadata key"); + String key = value.substring(0, equals); + checkArgument(!key.startsWith("GNU.sparse") && !key.equals("SCHILY.realsize") + && !key.equals("SCHILY.filetype"), "Sparse PAX archives are unsupported"); + offset += length; + } + } + } + + /// Counts actual reads, including chunked requests and decompressor-internal reads. + private static final class LimitedInputStream extends FilterInputStream { + private long _limit; + private long _read; + + private LimitedInputStream(InputStream input, long limit) { + super(input); + _limit = limit; + } + + private void limit(long limit) { + checkArgument(_read <= limit, "Multipart overhead exceeds its byte limit"); + _limit = limit; + } + + @Override + public int read() throws IOException { + int value = in.read(); + if (value != -1) { + add(1); + } + return value; + } + + @Override + public int read(byte[] bytes, int offset, int length) throws IOException { + int count = in.read(bytes, offset, (int) Math.min(length, _limit - _read + 1)); + if (count > 0) { + add(count); + } + return count; + } + + @Override + public long skip(long amount) throws IOException { + long skipped = 0; + byte[] buffer = new byte[(int) Math.min(8192, Math.max(0, amount))]; + while (skipped < amount) { + int count = read(buffer, 0, (int) Math.min(buffer.length, amount - skipped)); + if (count == -1) { + break; + } + skipped += count; + } + return skipped; + } + + private void add(int count) { + _read += count; + checkArgument(_read <= _limit, "Upload exceeds its byte limit"); + } + } +} diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeTaskPlacementStatus.java b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeTaskPlacementStatus.java index 7513d5823c1..30044a4cb01 100644 --- a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeTaskPlacementStatus.java +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/NativeTaskPlacementStatus.java @@ -58,8 +58,8 @@ final class NativeTaskPlacementStatus { BLOCKED, UNSUPPORTED, WAITING_FOR_SERVERS, UNDER_REPLICATED, INSTALLING, RECONCILING, INSTALLED, FAILED } - record Output(UUID taskId, String segmentName, UUID segmentObjectId, int requestedReplicas, int desiredReplicas, - int installedReplicas, int eligibleServers, State state, String detail) { + record Output(@Nullable UUID taskId, String segmentName, UUID segmentObjectId, int requestedReplicas, + int desiredReplicas, int installedReplicas, int eligibleServers, State state, String detail) { boolean needsReconciliation() { return state != State.BLOCKED && state != State.UNSUPPORTED && state != State.INSTALLED && state != State.FAILED; } @@ -104,7 +104,7 @@ final class NativeTaskPlacementStatus { _policy = policy; } - Output output(UUID taskId, Content content) throws IOException { + Output output(@Nullable UUID taskId, Content content) throws IOException { if (content.loader().tier() != null) { return result(taskId, content, -1, -1, -1, -1, State.UNSUPPORTED, "Tiered output placement is not supported"); @@ -249,8 +249,8 @@ final class NativeTaskPlacementStatus { return _records.getOrDefault(kind, Map.of()).get(name); } - private static Output result(UUID taskId, Content content, int requested, int desired, int installed, int eligible, - State state, String detail) { + private static Output result(@Nullable UUID taskId, Content content, int requested, int desired, int installed, + int eligible, State state, String detail) { return new Output(taskId, content.segmentName(), content.segmentObjectId(), requested, desired, installed, eligible, state, detail); } diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerRuntime.java b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerRuntime.java index 92683bcf735..d7979b10e03 100644 --- a/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerRuntime.java +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/coordination/RatisControllerRuntime.java @@ -196,6 +196,7 @@ public final class RatisControllerRuntime implements AutoCloseable, ServiceStatu _tasks = new NativeControllerTasks(requireNonNull(_access), requireNonNull(_access), _history, _instanceId, _config); _adminApp.register(new NativeControllerTaskResource(_tasks, accessControlFactory)); + _adminApp.register(new NativeControllerSegmentResource(metadata, _tasks, _config, accessControlFactory)); Instant startTime = Instant.now(); _adminApp.registerBinder(new AbstractBinder() { @Override diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerTasksTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerTasksTest.java index e19954a63d5..5a7ca836054 100644 --- a/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerTasksTest.java +++ b/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeControllerTasksTest.java @@ -289,6 +289,43 @@ public class NativeControllerTasksTest { } } + @Test + public void testDirectPublicationLostReplyAndRestartReconcileWithoutTask() throws Exception { + try (Fixture f = new Fixture(); NativeControllerTasks first = f.controller("first")) { + f.placementPolicy(true); + f.readyServer(); + var table = f.records(RecordKind.TABLE_CONFIG).getFirst(); + var schema = f.records(RecordKind.SCHEMA).getFirst(); + ObjectVersion tableVersion = new ObjectVersion(f._history, table.objectId(), table.version()); + ObjectVersion schemaVersion = new ObjectVersion(f._history, schema.objectId(), schema.version()); + Content content = new Content("orders_OFFLINE", table.objectId(), "uploaded", UUID.randomUUID(), + "a".repeat(64), "file:///approved/uploaded.tar.gz", + new OfflineSegmentProtocol.LoaderDescriptor(1, -1, false, null, null, 1, 1, Map.of())); + f._losePublicationReply = true; + expectThrows(IOException.class, () -> first.publish(content, tableVersion, schemaVersion)); + assertEquals(first.publish(content, tableVersion, schemaVersion).status(), Status.OK); + try (NativeControllerTasks other = f.controller("other")) { + assertEquals(other.publish(content, tableVersion, schemaVersion).status(), Status.OK, + "A different controller can resolve an identical committed publication without stealing ownership"); + } + first.tick(); + CatalogRecord assignment = f.records(RecordKind.OFFLINE_REPLICA_ASSIGNMENT).getFirst(); + assertEquals(f.records(RecordKind.OFFLINE_SEGMENT_CONTENT).size(), 1); + assertTrue(f.records(RecordKind.TASK).isEmpty(), "An HTTP upload must not manufacture a minion task"); + first.close(); + MetadataCatalog restored = new MetadataCatalog(f._cluster, f._history); + restored.restoreSnapshot(f._catalog.captureSnapshot()); + f._catalog = restored; + try (NativeControllerTasks replacement = f.controller("first")) { + assertEquals(replacement.publish(content, tableVersion, schemaVersion).status(), Status.OK); + replacement.tick(); + assertEquals(f.records(RecordKind.OFFLINE_REPLICA_ASSIGNMENT).getFirst(), assignment); + assertEquals(f.records(RecordKind.CONTROLLER_OPERATION).size(), 1); + assertEquals(f.records(RecordKind.OFFLINE_SEGMENT_CONTENT).size(), 1); + } + } + } + private static final class Fixture implements MetadataAccess, MembershipAccess, AutoCloseable { private final UUID _cluster = UUID.randomUUID(); private final UUID _history = UUID.randomUUID(); @@ -299,6 +336,7 @@ public class NativeControllerTasksTest { private boolean _loseRegistrationReply; private boolean _loseRenewalReply; private boolean _losePlacementReply; + private boolean _losePublicationReply; private int _taskCommands; private int _placementCommands; private long _workerSequence; @@ -456,7 +494,7 @@ public class NativeControllerTasksTest { throw new IOException("Lost response after committed schedule"); } if (command instanceof MetadataProtocol.OfflineSegmentCommand placement - && placement.request() instanceof OfflineSegmentProtocol.ReconcileTaskOutput) { + && placement.request() instanceof OfflineSegmentProtocol.ReconcileContent) { _placementCommands++; if (_losePlacementReply) { _losePlacementReply = false; @@ -464,6 +502,12 @@ public class NativeControllerTasksTest { throw new IOException("Lost response after committed placement"); } } + if (_losePublicationReply && command instanceof MetadataProtocol.OfflineSegmentCommand publication + && publication.request() instanceof OfflineSegmentProtocol.PublishValidatedContent) { + _losePublicationReply = false; + assertEquals(result.status(), Status.OK); + throw new IOException("Lost response after committed content publication"); + } return result; } diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeSegmentUploadStorageTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeSegmentUploadStorageTest.java new file mode 100644 index 00000000000..5d7c9924f9b --- /dev/null +++ b/pinot-controller/src/test/java/org/apache/pinot/controller/coordination/NativeSegmentUploadStorageTest.java @@ -0,0 +1,346 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.controller.coordination; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.FilterInputStream; +import java.io.IOException; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.List; +import java.util.UUID; +import org.apache.commons.compress.archivers.tar.TarArchiveEntry; +import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream; +import org.apache.commons.compress.archivers.tar.TarConstants; +import org.apache.commons.compress.compressors.gzip.GzipCompressorOutputStream; +import org.apache.commons.io.FileUtils; +import org.apache.pinot.common.utils.TarCompressionUtils; +import org.apache.pinot.controller.ControllerConf; +import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Content; +import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl; +import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader; +import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.data.FieldSpec.DataType; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.data.readers.GenericRow; +import org.apache.pinot.spi.utils.builder.TableConfigBuilder; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.expectThrows; + + +/// Real archives exercise immutable retries, destination-schema validation and bounds before catalog publication. +public class NativeSegmentUploadStorageTest { + private static final String BOUNDARY = "pinot-native-upload-test"; + private static final Schema SCHEMA = + new Schema.SchemaBuilder().setSchemaName("source").addSingleValueDimension("id", DataType.INT).build(); + private static final TableConfig TABLE = + new TableConfigBuilder(TableType.OFFLINE).setTableName("destination").build(); + + @Test + public void testDurableIdenticalRetryAcrossStorageInstancesAndDestinationSchema() throws Exception { + try (Fixture fixture = new Fixture()) { + byte[] archive = fixture.archive(); + Content first = fixture.store(archive, SCHEMA); + CloseAwareInput input = new CloseAwareInput(multipart(archive, false)); + Content repeated = new NativeSegmentUploadStorage(fixture._conf).store(input, BOUNDARY, + fixture._history, fixture._tableId, TABLE, SCHEMA); + assertEquals(repeated, first); + assertTrue(input._closed, "The outer request owner must close the entity after the final bounded read"); + assertEquals(first.tableName(), "destination_OFFLINE", "v2 tableName overrides embedded source table"); + assertEquals(first.loader().pushTimeMs(), -1L); + assertEquals(first.loader().dataCrc(), -1L); + assertFalse(first.loader().useDataCrc()); + assertEquals(Files.readAllBytes(Path.of(URI.create(first.downloadUri()))), archive); + Schema incompatible = + new Schema.SchemaBuilder().setSchemaName("destination").addSingleValueDimension("id", DataType.LONG).build(); + expectThrows(IllegalArgumentException.class, () -> fixture.store(archive, incompatible)); + fixture.noStaging(); + } + } + + @Test + public void testCrcAndExistingObjectCorruptionNeverPublishesOrOverwrites() throws Exception { + try (Fixture fixture = new Fixture()) { + byte[] archive = fixture.archive(); + Content content = fixture.store(archive, SCHEMA); + Path immutable = Path.of(URI.create(content.downloadUri())); + Files.writeString(immutable, "corrupted"); + expectThrows(IllegalArgumentException.class, () -> fixture.store(archive, SCHEMA)); + assertEquals(Files.readString(immutable), "corrupted"); + Path extra = fixture._directory.resolve("generated/segment/extra"); + Files.writeString(extra, "not covered by creation metadata"); + Path corruptArchive = fixture._directory.resolve("corrupt.tar.gz"); + TarCompressionUtils.createCompressedTarFile(extra.getParent().toFile(), corruptArchive.toFile()); + expectThrows(IllegalArgumentException.class, () -> fixture.store(Files.readAllBytes(corruptArchive), SCHEMA)); + fixture.noStaging(); + } + } + + @Test + public void testEscapedIncludesAndInterpolationAreRejectedBeforeMetadataEvaluation() throws Exception { + try (Fixture fixture = new Fixture()) { + fixture.archive(); + Path segment = fixture._directory.resolve("generated/segment"); + Path metadata; + try (var files = Files.walk(segment)) { + metadata = files.filter(path -> path.getFileName().toString().equals("metadata.properties")) + .findFirst().orElseThrow(); + } + byte[] original = Files.readAllBytes(metadata); + for (String property : List.of("inc\\u006cude = file:///native-upload-must-not-read", + "InClUdE\\u004fptional = file:///native-upload-must-not-read", + "custom.value = \\u0024{sys:user.home}", "custom.\\u0024{sys:user.home} = literal")) { + Files.write(metadata, original); + Files.writeString(metadata, "\n" + property + "\n", StandardOpenOption.APPEND); + Path archive = fixture._directory.resolve("properties.tar.gz"); + TarCompressionUtils.createCompressedTarFile(segment.toFile(), archive.toFile()); + IllegalArgumentException failure = expectThrows(IllegalArgumentException.class, + () -> fixture.store(Files.readAllBytes(archive), SCHEMA)); + assertTrue(failure.getMessage().contains("Segment properties must be literal")); + } + Files.write(metadata, original); + Path indexMap = segment.resolve("index_map"); + Files.writeString(indexMap, "inc\\u006cudeOptional = file:///native-upload-must-not-read"); + Path archive = fixture._directory.resolve("index-map.tar.gz"); + TarCompressionUtils.createCompressedTarFile(segment.toFile(), archive.toFile()); + assertTrue(expectThrows(IllegalArgumentException.class, + () -> fixture.store(Files.readAllBytes(archive), SCHEMA)).getMessage().contains("must be literal")); + fixture.noStaging(); + } + } + + @Test + public void testTraversalLinksDuplicatesAndExpansionBounds() throws Exception { + try (Fixture fixture = new Fixture()) { + for (String name : List.of("../outside", "segment/../../outside", "/outside", "segment\\outside")) { + byte[] archive = tar(new TarArchiveEntry(name), new byte[0]); + expectThrows(IllegalArgumentException.class, () -> fixture.store(archive, SCHEMA)); + } + TarArchiveEntry link = new TarArchiveEntry("segment/link", TarConstants.LF_SYMLINK); + link.setLinkName("/outside"); + expectThrows(IllegalArgumentException.class, () -> fixture.store(tar(link, new byte[0]), SCHEMA)); + byte[] duplicate = tar(new TarArchiveEntry("segment/duplicate"), new byte[0], + new TarArchiveEntry("segment/duplicate")); + expectThrows(IllegalArgumentException.class, () -> fixture.store(duplicate, SCHEMA)); + for (byte kind : List.of(TarConstants.LF_PAX_EXTENDED_HEADER_LC, TarConstants.LF_GNUTYPE_LONGNAME)) { + byte[] oversizedHeader = extension(kind, 1024L * 1024 * 1024, new byte[0], 1); + expectThrows(IllegalArgumentException.class, () -> fixture.store(oversizedHeader, SCHEMA)); + } + byte[] paxLength = "2147483647 path=x\n".getBytes(StandardCharsets.US_ASCII); + expectThrows(IllegalArgumentException.class, () -> fixture.store( + extension(TarConstants.LF_PAX_EXTENDED_HEADER_LC, paxLength.length, paxLength, 1), SCHEMA)); + expectThrows(IllegalArgumentException.class, () -> fixture.store( + extension(TarConstants.LF_GNUTYPE_LONGNAME, 2, new byte[]{'x', 0}, 5), SCHEMA)); + fixture._conf.setProperty(NativeSegmentUploadStorage.PREFIX + "maxExpandedBytes", 1024); + byte[] bomb = tar(new TarArchiveEntry("segment/large"), new byte[16 * 1024]); + expectThrows(IllegalArgumentException.class, () -> fixture.store(bomb, SCHEMA)); + fixture.noStaging(); + } + } + + @Test + public void testActualBodyLimitMultiplePartsAndSymlinkedDestination() throws Exception { + try (Fixture fixture = new Fixture()) { + byte[] archive = fixture.archive(); + fixture._conf.setProperty(NativeSegmentUploadStorage.PREFIX + "maxArchiveBytes", 32); + expectThrows(IllegalArgumentException.class, () -> fixture.store(archive, SCHEMA)); + fixture._conf.setProperty(NativeSegmentUploadStorage.PREFIX + "maxArchiveBytes", 128L * 1024 * 1024); + byte[] multipart = multipart(archive, true); + expectThrows(IllegalArgumentException.class, + () -> new NativeSegmentUploadStorage(fixture._conf).store(new ByteArrayInputStream(multipart), BOUNDARY, + fixture._history, fixture._tableId, TABLE, SCHEMA)); + for (boolean second : List.of(false, true)) { + byte[] header = oversizedMultipartHeader(archive, second); + expectThrows(IllegalArgumentException.class, + () -> new NativeSegmentUploadStorage(fixture._conf).store(new ByteArrayInputStream(header), BOUNDARY, + fixture._history, fixture._tableId, TABLE, SCHEMA)); + } + Path outside = Files.createDirectory(fixture._directory.resolve("outside")); + Files.createSymbolicLink(fixture._root.resolve(fixture._history.toString()), outside); + expectThrows(IllegalArgumentException.class, () -> fixture.store(archive, SCHEMA)); + try (var files = Files.list(outside)) { + assertEquals(files.count(), 0L); + } + fixture.noStaging(); + } + } + + @Test + public void testOrdinaryGnuLongSegmentNamesRemainSupported() throws Exception { + try (Fixture fixture = new Fixture()) { + String name = "segment_" + "a".repeat(150); + Content content = fixture.store(fixture.archive(name), SCHEMA); + assertEquals(content.segmentName(), name); + fixture.noStaging(); + } + } + + private static byte[] extension(byte kind, long size, byte[] payload, int count) throws Exception { + ByteArrayOutputStream output = new ByteArrayOutputStream(); + try (var gzip = new GzipCompressorOutputStream(output)) { + for (int i = 0; i < count; i++) { + TarArchiveEntry entry = new TarArchiveEntry("././@LongLink", kind); + entry.setSize(size); + byte[] header = new byte[512]; + entry.writeEntryHeader(header); + gzip.write(header); + gzip.write(payload); + gzip.write(new byte[(512 - payload.length % 512) % 512]); + } + gzip.write(new byte[1024]); + } + return output.toByteArray(); + } + + private static byte[] oversizedMultipartHeader(byte[] archive, boolean second) throws Exception { + ByteArrayOutputStream output = new ByteArrayOutputStream(); + output.write(("--" + BOUNDARY + "\r\nContent-Disposition: form-data; name=\"file\";" + + " filename=\"file.tar.gz\"\r\n").getBytes(StandardCharsets.US_ASCII)); + if (second) { + output.write("\r\n".getBytes(StandardCharsets.US_ASCII)); + output.write(archive); + output.write(("\r\n--" + BOUNDARY + "\r\n").getBytes(StandardCharsets.US_ASCII)); + } + output.write(("X-Oversized: " + "x".repeat(128 * 1024) + "\r\n\r\n").getBytes(StandardCharsets.US_ASCII)); + output.write(archive); + output.write(("\r\n--" + BOUNDARY + "--\r\n").getBytes(StandardCharsets.US_ASCII)); + return output.toByteArray(); + } + + private static byte[] tar(TarArchiveEntry first, byte[] bytes, TarArchiveEntry... more) throws Exception { + ByteArrayOutputStream output = new ByteArrayOutputStream(); + try (var gzip = new GzipCompressorOutputStream(output); var tar = new TarArchiveOutputStream(gzip)) { + first.setSize(bytes.length); + tar.putArchiveEntry(first); + tar.write(bytes); + tar.closeArchiveEntry(); + for (TarArchiveEntry entry : more) { + tar.putArchiveEntry(entry); + tar.closeArchiveEntry(); + } + } + return output.toByteArray(); + } + + private static byte[] multipart(byte[] archive, boolean extra) throws Exception { + ByteArrayOutputStream output = new ByteArrayOutputStream(); + output.write(("--" + BOUNDARY + "\r\nContent-Disposition: form-data; name=\"archive.tar.gz\";" + + " filename=\"../../ignored.tar.gz\"\r\nContent-Type: application/octet-stream\r\n\r\n") + .getBytes(StandardCharsets.US_ASCII)); + output.write(archive); + if (extra) { + output.write(("\r\n--" + BOUNDARY + + "\r\nContent-Disposition: form-data; name=\"extra\"\r\n\r\nextra").getBytes(StandardCharsets.US_ASCII)); + } + output.write(("\r\n--" + BOUNDARY + "--\r\n").getBytes(StandardCharsets.US_ASCII)); + return output.toByteArray(); + } + + private static final class Fixture implements AutoCloseable { + private final Path _directory = Files.createTempDirectory("native-segment-upload-"); + private final Path _root = Files.createDirectory(_directory.resolve("shared")); + private final ControllerConf _conf = new ControllerConf(); + private final UUID _history = UUID.randomUUID(); + private final UUID _tableId = UUID.randomUUID(); + + private Fixture() throws Exception { + _conf.setProperty(NativeSegmentUploadStorage.PREFIX + "storageRoot", _root.toString()); + } + + private byte[] archive() throws Exception { + return archive("segment"); + } + + private byte[] archive(String segmentName) throws Exception { + Path generated = Files.createDirectory(_directory.resolve("generated")); + SegmentGeneratorConfig config = new SegmentGeneratorConfig(TABLE, SCHEMA); + config.setTableName("source"); + config.setOutDir(generated.toString()); + config.setSegmentName(segmentName); + config.setCreationTime("123456789"); + SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl(); + GenericRow row = new GenericRow(); + row.putValue("id", 1); + driver.init(config, new GenericRowRecordReader(List.of(row))); + driver.build(); + Path archive = _directory.resolve("segment.tar.gz"); + TarCompressionUtils.createCompressedTarFile(generated.resolve(segmentName).toFile(), archive.toFile()); + return Files.readAllBytes(archive); + } + + private Content store(byte[] archive, Schema schema) throws Exception { + return new NativeSegmentUploadStorage(_conf).store(new ByteArrayInputStream(multipart(archive, false)), + BOUNDARY, _history, _tableId, TABLE, schema); + } + + private void noStaging() throws Exception { + try (var files = Files.list(_root)) { + assertTrue(files.noneMatch(path -> path.getFileName().toString().startsWith(".upload-"))); + } + } + + @Override + public void close() throws Exception { + FileUtils.deleteDirectory(_directory.toFile()); + } + } + + /// Model a request entity whose container rejects reads after close, unlike ByteArrayInputStream. + private static final class CloseAwareInput extends FilterInputStream { + private boolean _closed; + + private CloseAwareInput(byte[] bytes) { + super(new ByteArrayInputStream(bytes)); + } + + @Override + public int read() throws IOException { + checkOpen(); + return in.read(); + } + + @Override + public int read(byte[] bytes, int offset, int length) throws IOException { + checkOpen(); + return in.read(bytes, offset, length); + } + + @Override + public void close() throws IOException { + _closed = true; + super.close(); + } + + private void checkOpen() throws IOException { + if (_closed) { + throw new IOException("Request entity is closed"); + } + } + } +} diff --git a/pinot-coordination-ratis/CONTROLLER-API-PARITY.md b/pinot-coordination-ratis/CONTROLLER-API-PARITY.md index dbc10f40d07..f8e72652a06 100644 --- a/pinot-coordination-ratis/CONTROLLER-API-PARITY.md +++ b/pinot-coordination-ratis/CONTROLLER-API-PARITY.md @@ -27,14 +27,16 @@ The standard resource package has **264 annotated HTTP methods across 43 resourc ## Implementation status - Baseline: validated configuration reads/writes, controller authentication, HTTP/HTTPS, and metadata UI. -- Added in this change: shared table-size metadata adapter, optional read-only native server admin endpoint, +- Table sizes: shared table-size metadata adapter, optional read-only native server admin endpoint, identity-checked native OFFLINE table-size REST/UI support. Query listeners remain disabled. - Native minions: explicit `MINION_TASK` authorization, durable bounded schedules, normal minion startup and atomic output publication for `NativeSegmentGenerationTask`; see [the task contract](MINION-TASK-PLAN.md). -- Native task outputs now have bounded automatic replica placement from explicit administrator-owned tenant/domain - policy, durable installation intent and exact server receipts. The task API reports publication, desired and installed - replicas separately. This does not provide general rebalance, physical cleanup or query-serving qualification. -- Outstanding: general offline placement and lifecycle, full existing task API/plugin parity, realtime, +- Native immutable OFFLINE uploads use the normal multipart `POST /v2/segments` endpoint with captured table/schema + validation, durable content-addressed archives and fenced publication. Published content, including minion outputs, + has bounded automatic replica placement from explicit administrator-owned tenant/domain policy, durable installation + intent and exact server receipts. The task API reports publication, desired and installed replicas separately. + This does not provide replacement/deletion, general rebalance, physical cleanup or query-serving qualification. +- Outstanding: remaining offline lifecycle, full existing task API/plugin parity, realtime, remaining feature families and query-serving qualification. A full replacement is not yet implemented. - ZooKeeper tree APIs are not applicable to the replacement; a mutable compatibility tree will not be introduced. diff --git a/pinot-coordination-ratis/MINION-TASK-PLAN.md b/pinot-coordination-ratis/MINION-TASK-PLAN.md index 9162f8dab18..7562e2eab55 100644 --- a/pinot-coordination-ratis/MINION-TASK-PLAN.md +++ b/pinot-coordination-ratis/MINION-TASK-PLAN.md @@ -120,7 +120,8 @@ qualification remain outstanding. Input files default to 16 MiB and 100,000 rows of 64 MiB and 1,000,000 rows. Publication means committed content, not replica installation or query availability. The controller now reconciles -published task outputs through the existing native installation protocol. Its normal task scan shares one live +all published immutable OFFLINE content, including task outputs and validated controller uploads, through the existing +native installation protocol. Its normal task scan shares one live controller registration, table ownership and durable request stream with scheduling; each cycle attempts at most one schedule, retry or output reconciliation, rotating fairly over all candidates. Unknown replies retain their request identity. Controller replacement reclaims unfinished operations; completed current installations do not churn. diff --git a/pinot-coordination-ratis/README.md b/pinot-coordination-ratis/README.md index 2eedbb28331..c4d11a9dc64 100644 --- a/pinot-coordination-ratis/README.md +++ b/pinot-coordination-ratis/README.md @@ -281,9 +281,54 @@ its installation ID. Forced download bypasses local-byte reuse. Ordinary `LOAD` Publication refreshes work and checks local authority and the expected prior installation under the segment lock; consensus revalidates the receipt against later global changes. Neither local installation nor a committed receipt grants query-serving rights. This path currently supports ordinary OFFLINE tables with the default local segment -directory loader, without tiers, peer download, upsert or dimension tables. Controller REST -actions, controller work scheduling, routing publication and complete query-serving fences remain absent. This is an internal execution -path, not a complete controller/server reload integration or a supported Helix/ZooKeeper replacement. +directory loader, without tiers, peer download, upsert or dimension tables. Controller REST reload actions, +routing publication and complete query-serving fences remain outstanding. Automatic content reconciliation uses +this execution path; it is not a complete controller/server reload integration or a supported Helix/ZooKeeper replacement. + +## Immutable OFFLINE segment uploads + +The native controller accepts ordinary unencrypted segment archives through the normal multipart +`POST /v2/segments` endpoint. Configure an existing shared local directory on every controller: + +```properties +pinot.coordination.ratis.controller.upload.storageRoot=/absolute/shared-segments +``` + +The same directory must be accessible to the target servers; this slice uses `file:` archive locations and does not +implement a remote deep-store upload adapter. Configure the explicit placement policy described in +[MINION-TASK-PLAN.md](MINION-TASK-PLAN.md), then upload an existing segment: + +```bash +curl --fail-with-body -X POST \ + 'http://localhost:9000/v2/segments?tableName=baseballStats&tableType=OFFLINE&allowRefresh=false' \ + -F 'file=@/absolute/baseballStats-segment.tar.gz' +``` + +Use the deployment's normal controller credentials. Table-scoped upload authorization and database-name translation +apply before archive processing. The destination table and schema must already exist. Both captured versions and +the current table owner are checked atomically at publication. A concurrent configuration change rejects the upload +instead of publishing content validated against stale inputs. + +The controller bounds concurrent uploads, request/archive bytes, expanded bytes and archive entries, validates the +archive against the destination schema, and syncs the content-addressed archive before publishing its immutable +manifest. Retrying identical bytes produces the same content identity across controller restarts. Different bytes +under an existing segment name conflict. A successful response means committed publication; physical installation +is performed by reconciliation and measured separately by the table-size API. +Identical committed uploads can be resolved through any controller. New publication requires the table's live owner; +another controller returns an unavailable response rather than stealing that ownership. Owner forwarding remains +outstanding. The metadata UI does not yet expose an upload form. + +The component settings under `pinot.coordination.ratis.controller.upload.` are `maxArchiveBytes`, +`maxExpandedBytes` and `maxEntries`. They default to 128 MiB, 512 MiB and 4,096 respectively, with ceilings of +512 MiB, 2 GiB and 16,384. +Two uploads can be processed concurrently per controller. Multipart and tar extension metadata have separate +64 KiB bounds before parser allocation; these are not additional archive payload allowances. + +This endpoint supports one binary OFFLINE archive per request. Refresh/replacement, encrypted archives, URI/metadata +uploads, quota-configured tables, consistent batch push and batch uploads remain unsupported. Native query listeners +remain disabled. +Unreferenced finalized archives can remain after rejected or unknown publication; reclamation requires a separate +durable lifecycle protocol. ## Query identity and pinned content @@ -550,7 +595,7 @@ Validate the shared query wire context and deadline/permit composition: ``` The native minion execution slice adds bounded durable schedules, the normal `StartMinion` lifecycle and fenced -publication for actual OFFLINE JSON-lines segment generation. Published task outputs can be placed automatically +publication for actual OFFLINE JSON-lines segment generation. Published immutable OFFLINE content can be placed automatically using the explicit cluster key `pinot.coordination.ratis.offline.placement`. This strict policy maps approved stable SERVER IDs to tenant and failure-domain labels; no participant advertisement supplies placement authority. The controller reconciles desired replicas, installation operations and exact current server receipts while preserving diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/TaskOutputPlacementCatalog.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineContentPlacementCatalog.java similarity index 89% rename from pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/TaskOutputPlacementCatalog.java rename to pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineContentPlacementCatalog.java index 56a35ecdc19..c8f92641cfd 100644 --- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/TaskOutputPlacementCatalog.java +++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineContentPlacementCatalog.java @@ -42,12 +42,14 @@ import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Assig import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Assignment; import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Compatibility; import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Content; +import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.ReconcileContent; import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.ReconcileTaskOutput; import org.apache.pinot.coordination.ratis.metadata.OperationProtocol.AssignedOfflineSegmentReload; import org.apache.pinot.coordination.ratis.metadata.OperationProtocol.OfflineSegmentReload; import org.apache.pinot.coordination.ratis.metadata.OperationProtocol.OperationState; import org.apache.pinot.coordination.ratis.metadata.OperationProtocol.State; import org.apache.pinot.coordination.ratis.metadata.OperationProtocol.TargetMaterialization; +import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol.OwnerToken; import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.PublicationState; import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.TaskState; import org.apache.pinot.coordination.ratis.metadata.TaskProtocol.TaskStatus; @@ -56,10 +58,11 @@ import org.apache.pinot.spi.utils.JsonUtils; import org.apache.pinot.spi.utils.builder.TableNameBuilder; -/// Reconciles one published task output against one committed image. New assignments and their installation intent -/// commit together; subsequent commands claim and complete that intent. No external I/O or clocks affect placement. -final class TaskOutputPlacementCatalog { - private TaskOutputPlacementCatalog() { +/// Reconciles one published immutable OFFLINE segment against one committed image. New assignments and their +/// installation intent commit together; subsequent commands claim and complete that intent. No external I/O or +/// clocks affect placement. +final class OfflineContentPlacementCatalog { + private OfflineContentPlacementCatalog() { } static List<CatalogRecord> stage(ReconcileTaskOutput request, OperationCatalog.View view, long index) @@ -79,7 +82,20 @@ final class TaskOutputPlacementCatalog { "Segment is not an immutable output of this task publication"); require(content.loader().tier() == null, Status.INVALID_REQUEST, "Native task output placement does not support tiered content"); - OwnershipCatalog.validateTableOwner(request.owner(), content.tableName(), view); + return reconcile(request.owner(), content, view, index); + } + + static List<CatalogRecord> stage(ReconcileContent request, OperationCatalog.View view, long index) + throws IOException { + Content content = OfflineSegmentCatalog.currentContent(request.segmentObjectId(), view); + require(content.loader().tier() == null, Status.INVALID_REQUEST, + "Native content placement does not support tiered content"); + return reconcile(request.owner(), content, view, index); + } + + private static List<CatalogRecord> reconcile(OwnerToken owner, Content content, OperationCatalog.View view, + long index) throws IOException { + OwnershipCatalog.validateTableOwner(owner, content.tableName(), view); CatalogRecord tableRecord = view.record(content.tableObjectId()); CatalogRecord schemaRecord = view.record(RecordKind.SCHEMA, TableNameBuilder.extractRawTableName(content.tableName())); @@ -131,7 +147,7 @@ final class TaskOutputPlacementCatalog { for (var entry : previous.entrySet()) { Assignment assignment = entry.getValue(); if (assignment.desiredOnline() && !selected.contains(entry.getKey())) { - staged.add(OfflineSegmentCatalog.stage(new AssignReplica(request.owner(), content.segmentObjectId(), + staged.add(OfflineSegmentCatalog.stage(new AssignReplica(owner, content.segmentObjectId(), assignment.assignmentEpoch(), assignment.server(), assignment.compatibility(), false), staged, index)); } } @@ -140,7 +156,7 @@ final class TaskOutputPlacementCatalog { ParticipantState server = eligible.get(id); if (assignment == null || !assignment.desiredOnline() || !assignment.server().equals(server.participant()) || !assignment.compatibility().equals(compatibility)) { - staged.add(OfflineSegmentCatalog.stage(new AssignReplica(request.owner(), content.segmentObjectId(), + staged.add(OfflineSegmentCatalog.stage(new AssignReplica(owner, content.segmentObjectId(), assignment == null ? 0 : assignment.assignmentEpoch(), server.participant(), compatibility, true), staged, index)); increased.add(id); @@ -158,13 +174,13 @@ final class TaskOutputPlacementCatalog { OperationCatalog.currentAssigned(assigned, staged); } catch (CoordinationRejection e) { staged.add(OperationCatalog.stageNative(new OperationProtocol.Cancel(operation.operationId(), - record.version(), request.owner(), "Replica assignment superseded"), staged, index)); + record.version(), owner, "Replica assignment superseded"), staged, index)); } } for (String id : selected.stream().sorted().toList()) { Assignment assignment = OfflineSegmentCatalog.currentAssignment(content.segmentObjectId(), eligible.get(id).participant(), staged); - if (reconcileOperation(request, content, assignment, staged, index)) { + if (reconcileOperation(owner, content, assignment, staged, index)) { increased.add(id); } } @@ -174,7 +190,7 @@ final class TaskOutputPlacementCatalog { return List.copyOf(staged._staged.values()); } - private static boolean reconcileOperation(ReconcileTaskOutput request, Content content, Assignment assignment, + private static boolean reconcileOperation(OwnerToken owner, Content content, Assignment assignment, Overlay view, long index) throws IOException { UUID operationId = NativeReplicaPlacementPolicy.operationId(view.historyId(), assignment.materializationId()); CatalogRecord record = view.record(RecordKind.CONTROLLER_OPERATION, OperationProtocol.recordName(operationId)); @@ -193,15 +209,15 @@ final class TaskOutputPlacementCatalog { // A terminal failure is visible for operator intervention; never manufacture endless new epochs or retries. return false; } - if (operation.status() == State.ACCEPTED || !operation.owner().equals(request.owner())) { + if (operation.status() == State.ACCEPTED || !operation.owner().equals(owner)) { AssignedOfflineSegmentReload assigned = (AssignedOfflineSegmentReload) operation.input(); - view.add(OperationCatalog.stageNative(new OperationProtocol.Claim(operationId, record.version(), request.owner(), + view.add(OperationCatalog.stageNative(new OperationProtocol.Claim(operationId, record.version(), owner, assigned.dependencies()), view, index)); return false; } try { view.add(OperationCatalog.stageNative(new OperationProtocol.CompleteInstalled(operationId, record.version(), - request.owner(), operation.attempt()), view, index)); + owner, operation.attempt()), view, index)); } catch (CoordinationRejection e) { if (e.status() != Status.NOT_READY) { throw e; diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineSegmentCatalog.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineSegmentCatalog.java index b909c55c5d8..72f3ad309a7 100644 --- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineSegmentCatalog.java +++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineSegmentCatalog.java @@ -40,6 +40,8 @@ import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Conte import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.LoaderDescriptor; import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Operation; import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.PublishContent; +import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.PublishValidatedContent; +import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.ReconcileContent; import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.ReconcileTaskOutput; import org.apache.pinot.spi.config.table.TableType; import org.apache.pinot.spi.utils.builder.TableNameBuilder; @@ -61,8 +63,10 @@ final class OfflineSegmentCatalog { "Offline operation and committed index are required"); return switch (operation) { case PublishContent publish -> publish(publish, view, index); + case PublishValidatedContent publish -> publishValidated(publish, view, index); case AssignReplica assign -> assign(assign, view, index); - case ReconcileTaskOutput reconcile -> TaskOutputPlacementCatalog.stage(reconcile, view, index); + case ReconcileTaskOutput reconcile -> OfflineContentPlacementCatalog.stage(reconcile, view, index); + case ReconcileContent reconcile -> OfflineContentPlacementCatalog.stage(reconcile, view, index); }; } @@ -83,6 +87,30 @@ final class OfflineSegmentCatalog { MetadataProtocol.FORMAT_VERSION, index, false, MetadataProtocol.toTree(content))); } + private static List<CatalogRecord> publishValidated(PublishValidatedContent request, View view, long index) + throws IOException { + Content content = request.content(); + validateContent(content); + OwnershipCatalog.validateTableOwner(request.owner(), content.tableName(), view); + Compatibility compatibility = new Compatibility(request.expectedTable(), request.expectedSchema()); + validateCompatibility(compatibility, content.tableObjectId(), view.historyId()); + currentCompatibility(compatibility, content, view); + String name = OfflineSegmentProtocol.contentName(content.tableObjectId(), content.segmentName()); + CatalogRecord named = view.record(RecordKind.OFFLINE_SEGMENT_CONTENT, name); + CatalogRecord identified = view.record(content.segmentObjectId()); + if (named != null || identified != null) { + require(named != null && identified != null && !named.deleted() && !identified.deleted() + && named.objectId().equals(content.segmentObjectId()) + && identified.kind() == RecordKind.OFFLINE_SEGMENT_CONTENT + && identified.name().equals(name) && content.equals(OfflineSegmentProtocol.decodeContent(named.value())) + && content.equals(OfflineSegmentProtocol.decodeContent(identified.value())), Status.CONFLICT, + "Offline content name or object ID already identifies a different immutable manifest"); + return List.of(); + } + return List.of(new CatalogRecord(RecordKind.OFFLINE_SEGMENT_CONTENT, name, content.segmentObjectId(), 1, + MetadataProtocol.FORMAT_VERSION, index, false, MetadataProtocol.toTree(content))); + } + private static List<CatalogRecord> assign(AssignReplica request, View view, long index) throws IOException { require(request.segmentObjectId() != null && request.expectedEpoch() >= 0, Status.INVALID_REQUEST, diff --git a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineSegmentProtocol.java b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineSegmentProtocol.java index 10665e8d6ad..d3a9640d592 100644 --- a/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineSegmentProtocol.java +++ b/pinot-coordination-ratis/src/main/java/org/apache/pinot/coordination/ratis/metadata/OfflineSegmentProtocol.java @@ -72,14 +72,23 @@ public final class OfflineSegmentProtocol { @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "operation") @JsonSubTypes({@JsonSubTypes.Type(value = PublishContent.class, name = "PUBLISH_CONTENT"), @JsonSubTypes.Type(value = AssignReplica.class, name = "ASSIGN_REPLICA"), - @JsonSubTypes.Type(value = ReconcileTaskOutput.class, name = "RECONCILE_TASK_OUTPUT")}) - public sealed interface Operation permits PublishContent, AssignReplica, ReconcileTaskOutput { + @JsonSubTypes.Type(value = ReconcileTaskOutput.class, name = "RECONCILE_TASK_OUTPUT"), + @JsonSubTypes.Type(value = PublishValidatedContent.class, name = "PUBLISH_VALIDATED_CONTENT"), + @JsonSubTypes.Type(value = ReconcileContent.class, name = "RECONCILE_CONTENT")}) + public sealed interface Operation + permits PublishContent, AssignReplica, ReconcileTaskOutput, PublishValidatedContent, ReconcileContent { OwnerToken owner(); } public record PublishContent(OwnerToken owner, ObjectVersion expectedTable, Content content) implements Operation { } + /// Publishes an externally validated immutable archive only while both exact validation dependencies remain + /// current. An identical existing manifest is a no-op; the state machine does not fetch or inspect archive bytes. + public record PublishValidatedContent(OwnerToken owner, ObjectVersion expectedTable, ObjectVersion expectedSchema, + Content content) implements Operation { + } + /// An OFFLINE release requires the exact previous server token and compatibility, even when that registration or /// table incarnation is no longer current. ONLINE assignment requires live current dependencies and a ready server. public record AssignReplica(OwnerToken owner, UUID segmentObjectId, long expectedEpoch, ParticipantToken server, @@ -91,6 +100,10 @@ public final class OfflineSegmentProtocol { public record ReconcileTaskOutput(OwnerToken owner, UUID taskId, UUID segmentObjectId) implements Operation { } + /// Reconciles already published immutable content, independently of whether a task or validated upload produced it. + public record ReconcileContent(OwnerToken owner, UUID segmentObjectId) implements Operation { + } + public static Content decodeContent(JsonNode value) throws IOException { return MetadataProtocol.fromTree(value, Content.class); diff --git a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalogTest.java b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalogTest.java index 47c8ee3a3c2..ccbec5da657 100644 --- a/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalogTest.java +++ b/pinot-coordination-ratis/src/test/java/org/apache/pinot/coordination/ratis/metadata/TaskCatalogTest.java @@ -68,6 +68,8 @@ import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Assig import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.Content; import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.LoaderDescriptor; import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.PublishContent; +import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.PublishValidatedContent; +import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.ReconcileContent; import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol.ReconcileTaskOutput; import org.apache.pinot.coordination.ratis.metadata.OperationProtocol.OperationState; import org.apache.pinot.coordination.ratis.metadata.OperationProtocol.State; @@ -641,6 +643,84 @@ public class TaskCatalogTest { f.restore(); } + @Test + public void testValidatedPublicationFencesSchemaTableAndImmutableRetry() throws Exception { + Fixture f = new Fixture(true); + Content content = f.output(f._task, 1); + ObjectVersion table = f.objectVersion(f.record(RecordKind.TABLE_CONFIG, TABLE)); + ObjectVersion schema = f.objectVersion(f.record(RecordKind.SCHEMA, "orders")); + PublishValidatedContent original = new PublishValidatedContent(f._owner, table, schema, content); + Schema changedSchema = new Schema.SchemaBuilder().setSchemaName("orders") + .addSingleValueDimension("id", DataType.INT).addSingleValueDimension("extra", DataType.STRING).build(); + assertEquals(f.mutate(Mutation.updateSchema(schema, changedSchema)).status(), Status.OK); + assertEquals(f.segment(original).status(), Status.CONFLICT, + "Schema changes after validation must fence publication"); + schema = f.objectVersion(f.record(RecordKind.SCHEMA, "orders")); + assertEquals(f.mutate(Mutation.updateTableConfig(table, f.table())).status(), Status.OK); + assertEquals(f.segment(new PublishValidatedContent(f._owner, table, schema, content)).status(), Status.CONFLICT, + "Table changes after validation must fence publication"); + table = f.objectVersion(f.record(RecordKind.TABLE_CONFIG, TABLE)); + UUID unrelated = UUID.randomUUID(); + assertEquals(f.mutate(Mutation.createSchema(f._history, unrelated, new Schema.SchemaBuilder().setSchemaName("other") + .addSingleValueDimension("id", DataType.INT).build())).status(), Status.OK); + assertEquals(f.segment(new PublishValidatedContent(f._owner, table, new ObjectVersion(f._history, unrelated, 1), + content)).status(), Status.CONFLICT, "A different active schema cannot validate this table's upload"); + PublishValidatedContent current = new PublishValidatedContent(f._owner, table, schema, content); + Result published = f.segment(current); + assertEquals(published.status(), Status.OK); + assertEquals(published.versions().size(), 1); + CatalogRecord immutable = f.record(RecordKind.OFFLINE_SEGMENT_CONTENT, + OfflineSegmentProtocol.contentName(content.tableObjectId(), content.segmentName())); + assertEquals(f.segment(current).status(), Status.OK); + assertTrue(f.segment(current).versions().isEmpty(), "An exact retry must not advance immutable content"); + assertEquals(f.record(RecordKind.OFFLINE_SEGMENT_CONTENT, immutable.name()), immutable); + for (Content conflict : List.of( + new Content(TABLE, f._table, content.segmentName(), content.segmentObjectId(), "b".repeat(64), + content.downloadUri(), content.loader()), + new Content(TABLE, f._table, content.segmentName(), UUID.randomUUID(), content.contentSha256(), + content.downloadUri(), content.loader()), + new Content(TABLE, f._table, "other-segment", content.segmentObjectId(), content.contentSha256(), + content.downloadUri(), content.loader()))) { + assertEquals(f.segment(new PublishValidatedContent(f._owner, table, schema, conflict)).status(), Status.CONFLICT); + } + assertEquals(f.segment(new PublishContent(f._owner, table, content)).status(), Status.CONFLICT, + "The legacy publication command must retain its original duplicate behavior"); + f.restore(); + assertTrue(f.segment(current).versions().isEmpty()); + assertEquals(f.record(RecordKind.OFFLINE_SEGMENT_CONTENT, immutable.name()), immutable); + } + + @Test + public void testValidatedContentAutomaticallyReconcilesWithoutTaskRecords() throws Exception { + Fixture f = new Fixture(); + Content content = new Content(TABLE, f._table, "uploaded-segment", UUID.randomUUID(), "c".repeat(64), + "file:///uploads/immutable.tar.gz", new LoaderDescriptor(1, 0, false, null, null, 1, 1, Map.of())); + assertEquals(f.segment(new PublishValidatedContent(f._owner, + f.objectVersion(f.record(RecordKind.TABLE_CONFIG, TABLE)), + f.objectVersion(f.record(RecordKind.SCHEMA, "orders")), content)).status(), Status.OK); + f.register("server", Role.SERVER, 0); + f.placementPolicy(Map.of("server", new NativeReplicaPlacementPolicy.Server("DefaultTenant", "zone-a"))); + ReconcileContent reconcile = new ReconcileContent(f._owner, content.segmentObjectId()); + Result accepted = f.segment(reconcile); + assertEquals(accepted.status(), Status.OK); + assertEquals(accepted.versions().size(), 2, "Direct content also atomically creates assignment and install intent"); + Assignment assignment = f.assignments(content).getFirst(); + assertEquals(f.operation(assignment).status(), State.ACCEPTED); + assertEquals(f.segment(new ReconcileTaskOutput(f._owner, f._task, content.segmentObjectId())).status(), + Status.NOT_FOUND, "The task-specific command must still require a successful task publication"); + f.restore(); + assertEquals(f.segment(reconcile).status(), Status.OK); + assertEquals(f.operation(assignment).status(), State.RUNNING); + f.install(assignment); + assertEquals(f.segment(reconcile).status(), Status.OK); + assertEquals(f.operation(assignment).status(), State.SUCCEEDED); + assertEquals(f.operation(assignment).completionProof().installations().size(), 1); + assertTrue(f.segment(reconcile).versions().isEmpty()); + assertTrue(f._catalog.captureImage(new FeedScope(Set.of(RecordKind.TASK, RecordKind.TASK_PUBLICATION), "no-task")) + .records().isEmpty(), "Direct publication must not manufacture task records"); + f.restore(); + } + private static final class Fixture { private final boolean _nativeTask; private final UUID _cluster = UUID.randomUUID(); diff --git a/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java b/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java index 570ccae877c..c5c28420c8e 100644 --- a/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java +++ b/pinot-tools/src/test/java/org/apache/pinot/tools/service/RatisStartersTest.java @@ -68,12 +68,10 @@ import org.apache.pinot.coordination.ratis.metadata.NativeReplicaPlacementPolicy import org.apache.pinot.coordination.ratis.metadata.OfflineInstallationProtocol; import org.apache.pinot.coordination.ratis.metadata.OfflineSegmentProtocol; import org.apache.pinot.coordination.ratis.metadata.OperationProtocol; -import org.apache.pinot.coordination.ratis.metadata.OwnershipProtocol; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc; import org.apache.pinot.coordination.ratis.rpc.MetadataRpc.ResultStatus; import org.apache.pinot.coordination.ratis.transport.ControllerAccessPolicy; import org.apache.pinot.coordination.ratis.transport.ControllerAccessPolicy.ControllerIdentity; -import org.apache.pinot.coordination.ratis.transport.MetadataGrpcClient; import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy; import org.apache.pinot.coordination.ratis.transport.ParticipantAccessPolicy.ParticipantIdentity; import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl; @@ -181,6 +179,7 @@ public class RatisStartersTest { Files.writeString(trust, Files.readString(controller.certificate()) + Files.readString(broker.certificate()) + Files.readString(server.certificate()) + Files.readString(minion.certificate())); Path taskOutput = Files.createDirectories(directory.resolve("task-output")); + Path uploadStorage = Files.createDirectories(directory.resolve("upload-storage")); int[] ports = ports(12); String history = UUID.randomUUID().toString(); String cluster = UUID.randomUUID().toString(); @@ -223,6 +222,7 @@ public class RatisStartersTest { config.put("test.minionFingerprint", minion.fingerprint()); config.put(PREFIX + "controller.tasks.outputRoot", taskOutput.toString()); config.put(PREFIX + "controller.tasks.pollMillis", 100L); + config.put(PREFIX + "controller.upload.storageRoot", uploadStorage.toString()); controllers.add(config); starts.add(starters.submit(() -> manager.startRole(ServiceRole.CONTROLLER, config))); } @@ -393,8 +393,17 @@ public class RatisStartersTest { updated = write(http, controllerUrl + "/tables/orders_OFFLINE", "PUT", JsonUtils.objectToString(table), Map.of("If-Match", tableTag)); assertEquals(updated.statusCode(), 200, updated.body()); - long expectedBytes = installSizeTestSegment(directory, new PinotConfiguration(controllers.getFirst()), - metadata, serverState.getParticipant(), table, schema); + String placementPolicy = JsonUtils.objectToString(Map.of("servers", Map.of(SERVER, + Map.of("tenant", "DefaultTenant", "failureDomain", "zone-a")))); + clusterTag = get(http, controllerUrl + "/cluster/configs", true).headers().firstValue("ETag").orElseThrow(); + updated = write(http, controllerUrl + "/cluster/configs", "POST", + JsonUtils.objectToString(Map.of(PREFIX + "offline.placement", placementPolicy)), + Map.of("If-Match", clusterTag)); + assertEquals(updated.statusCode(), 200, updated.body()); + PinotConfiguration voterConfig = new PinotConfiguration(controllers.getFirst()); + InstalledOutput uploaded = uploadSizeTestSegment(directory, voterConfig, http, controllerUrl, + secondControllerUrl, serverState.getParticipant(), table, schema); + long expectedBytes = uploaded.bytes(); awaitSize(http, serverAdmin + "/tables/orders_OFFLINE/size", "diskSizeInBytes", expectedBytes); assertMissingSize(http, controllerUrl); String approvalKey = PREFIX + "server.admin.endpoint." + SERVER; @@ -420,9 +429,8 @@ public class RatisStartersTest { awaitSize(http, controllerUrl + "/tables/orders/size", "reportedSizeInBytes", expectedBytes); assertEquals(ServiceStatus.getServiceStatus(BROKER), ServiceStatus.Status.BAD); assertEquals(ServiceStatus.getServiceStatus(SERVER), ServiceStatus.Status.GOOD); - PinotConfiguration voterConfig = new PinotConfiguration(controllers.getFirst()); - TaskOutput generatedOutput = runNativeTask(manager, directory, history, endpoints, minion, controller, http, - controllerUrl, taskOutput, voterConfig, serverState.getParticipant(), serverAdmin); + InstalledOutput generatedOutput = runNativeTask(manager, directory, history, endpoints, minion, controller, + http, controllerUrl, taskOutput, voterConfig, serverState.getParticipant(), serverAdmin); for (int i = 6; i < 8; i++) { try (ServerSocket unused = new ServerSocket(ports[i])) { assertTrue(unused.isBound(), "Native query listener must remain closed"); @@ -438,8 +446,15 @@ public class RatisStartersTest { MetadataRpc.ParticipantToken replacement = serverClient.membershipRead(read).getParticipant().getParticipant(); assertEquals(replacement.getRegistrationEpoch(), serverState.getParticipant().getRegistrationEpoch() + 1); assertNotEquals(replacement.getIncarnation(), serverState.getParticipant().getIncarnation()); - assertMissingSize(http, controllerUrl); // Unmanaged content cannot borrow the replacement process's authority. - TaskInstallation reinstalled = awaitTaskInstallation(voterConfig, replacement, generatedOutput.content()); + ReplicaInstallation uploadedAgain = awaitInstallation(voterConfig, replacement, uploaded.content()); + assertTrue(uploadedAgain.assignment().assignmentEpoch() + > uploaded.installation().assignment().assignmentEpoch()); + assertNotEquals(uploadedAgain.assignment().materializationId(), + uploaded.installation().assignment().materializationId()); + assertNotEquals(uploadedAgain.installationId(), uploaded.installation().installationId()); + awaitSize(http, serverAdmin + "/tables/orders_OFFLINE/size", "diskSizeInBytes", expectedBytes); + awaitSize(http, controllerUrl + "/tables/orders/size", "reportedSizeInBytes", expectedBytes); + ReplicaInstallation reinstalled = awaitInstallation(voterConfig, replacement, generatedOutput.content()); assertTrue(reinstalled.assignment().assignmentEpoch() > generatedOutput.installation().assignment() .assignmentEpoch()); assertNotEquals(reinstalled.assignment().materializationId(), @@ -460,8 +475,10 @@ public class RatisStartersTest { assertEquals(manager.startRole(ServiceRole.CONTROLLER, controllers.getFirst()), "Controller_ratis_voter-0"); assertEquals(metadata.membershipRead(read).getControl().getPolicy().getPermitLifetimeMillis(), 1000L, "Restart must retain committed policy instead of resetting it from bootstrap configuration"); - TaskInstallation recovered = awaitTaskInstallation(voterConfig, replacement, generatedOutput.content()); + ReplicaInstallation recovered = awaitInstallation(voterConfig, replacement, generatedOutput.content()); assertEquals(recovered, reinstalled, "Controller restart must retain the current installation and proof"); + assertEquals(awaitInstallation(voterConfig, replacement, uploaded.content()), uploadedAgain, + "Controller restart must not create another materialization for an uploaded segment"); var taskState = JsonUtils.stringToJsonNode(get(http, controllerUrl + "/tasks/native/nativeTasks", true).body()); assertEquals(taskState.path("tasks").size(), 1); assertEquals(taskState.path("publications").size(), 1); @@ -499,23 +516,17 @@ public class RatisStartersTest { } } - private record TaskInstallation(OfflineSegmentProtocol.Assignment assignment, + private record ReplicaInstallation(OfflineSegmentProtocol.Assignment assignment, OperationProtocol.OperationState operation, UUID installationId) { } - private record TaskOutput(OfflineSegmentProtocol.Content content, long bytes, TaskInstallation installation) { + private record InstalledOutput(OfflineSegmentProtocol.Content content, long bytes, ReplicaInstallation installation) { } - private static TaskOutput runNativeTask(PinotServiceManager manager, Path directory, String history, String endpoints, - CertificateFiles minion, CertificateFiles controller, HttpClient http, String controllerUrl, Path output, - PinotConfiguration voterConfig, MetadataRpc.ParticipantToken server, String serverAdmin) throws Exception { - String clusterTag = get(http, controllerUrl + "/cluster/configs", true).headers().firstValue("ETag").orElseThrow(); - String placementPolicy = JsonUtils.objectToString(Map.of("servers", Map.of(SERVER, - Map.of("tenant", "DefaultTenant", "failureDomain", "zone-a")))); - HttpResponse<String> configured = write(http, controllerUrl + "/cluster/configs", "POST", - JsonUtils.objectToString(Map.of(PREFIX + "offline.placement", placementPolicy)), - Map.of("If-Match", clusterTag)); - assertEquals(configured.statusCode(), 200, configured.body()); + private static InstalledOutput runNativeTask(PinotServiceManager manager, Path directory, String history, + String endpoints, CertificateFiles minion, CertificateFiles controller, HttpClient http, String controllerUrl, + Path output, PinotConfiguration voterConfig, MetadataRpc.ParticipantToken server, String serverAdmin) + throws Exception { Path sourceRoot = Files.createDirectories(directory.resolve("task-source")); Path source = sourceRoot.resolve("rows.json"); Files.writeString(source, "{\"id\":1}\n{\"id\":2}\n"); @@ -578,7 +589,7 @@ public class RatisStartersTest { .toFile()); assertEquals(generated.getTotalDocs(), 2); OfflineSegmentProtocol.Content published = OfflineSegmentProtocol.decodeContent(content); - TaskInstallation installation = awaitTaskInstallation(voterConfig, server, published); + ReplicaInstallation installation = awaitInstallation(voterConfig, server, published); long bytes = FileUtils.sizeOfDirectory(unpacked.resolve(content.path("segmentName").asText()).resolve("v3") .toFile()); awaitSize(http, serverAdmin + "/tables/nativeTasks_OFFLINE/size", "diskSizeInBytes", bytes); @@ -589,14 +600,14 @@ public class RatisStartersTest { snapshot = JsonUtils.stringToJsonNode(get(http, url, true).body()); assertEquals(snapshot.path("tasks").size(), 1); assertEquals(snapshot.path("publications").size(), 1); - return new TaskOutput(published, bytes, installation); + return new InstalledOutput(published, bytes, installation); } finally { manager.stopPinotInstanceById("Minion_native_test"); } } /// Reads committed placement and receipt proof; no test client submits assignments or installation operations. - private static TaskInstallation awaitTaskInstallation(PinotConfiguration config, + private static ReplicaInstallation awaitInstallation(PinotConfiguration config, MetadataRpc.ParticipantToken serverSession, OfflineSegmentProtocol.Content content) throws Exception { UUID history = UUID.fromString(serverSession.getHistoryId()); var server = new MembershipProtocol.ParticipantToken(history, serverSession.getParticipantId(), Role.SERVER, @@ -628,7 +639,7 @@ public class RatisStartersTest { assertEquals(receipt.installed().segmentObjectId(), content.segmentObjectId()); assertEquals(receipt.installed().assignment().version(), assignment.assignmentEpoch()); assertEquals(receipt.installed().materializationId(), assignment.materializationId()); - return new TaskInstallation(assignment, operation, receipt.installationId()); + return new ReplicaInstallation(assignment, operation, receipt.installationId()); } } } @@ -638,9 +649,60 @@ public class RatisStartersTest { } } - /// Uses the existing trusted consensus test API to publish content; public segment upload remains a separate API. - private static long installSizeTestSegment(Path directory, PinotConfiguration config, MetadataRpcClient metadata, - MetadataRpc.ParticipantToken serverSession, TableConfig table, Schema schema) throws Exception { + /// Uploads a real archive through the normal API and observes automatic installation without metadata mutations. + private static InstalledOutput uploadSizeTestSegment(Path directory, PinotConfiguration config, HttpClient http, + String controllerUrl, String secondControllerUrl, MetadataRpc.ParticipantToken serverSession, TableConfig table, + Schema schema) throws Exception { + PreparedSegment prepared = prepareSizeTestSegment(directory.resolve("upload-source"), table, schema, 2L); + HttpResponse<String> response = upload(http, controllerUrl, table.getTableName(), prepared.archive(), false); + assertEquals(response.statusCode(), 401, response.body()); + for (String condition : List.of("If-Match", "If-None-Match")) { + response = upload(http, controllerUrl, table.getTableName(), prepared.archive(), true, Map.of(condition, "*")); + assertEquals(response.statusCode(), 400, response.body()); + assertTrue(JsonUtils.stringToJsonNode(response.body()).path("error").asText() + .contains("Unsupported native upload option"), response.body()); + } + response = upload(http, controllerUrl, table.getTableName(), prepared.archive(), true, + Map.of(CommonConstants.DATABASE, "default", CommonConstants.Controller.TABLE_NAME_HTTP_HEADER, + "other.orders_OFFLINE")); + assertEquals(response.statusCode(), 400, response.body()); + assertTrue(JsonUtils.stringToJsonNode(response.body()).path("error").asText().contains("Invalid table name"), + response.body()); + response = upload(http, controllerUrl, table.getTableName(), prepared.archive(), true); + assertEquals(response.statusCode(), 200, response.body()); + UUID history = UUID.fromString(serverSession.getHistoryId()); + try (RatisMetadataStore.Client consensus = RatisMetadataStore.connect(config)) { + CatalogRecord tableRecord = record(consensus, history, MetadataProtocol.RecordKind.TABLE_CONFIG, + table.getTableName()); + String contentName = OfflineSegmentProtocol.contentName(tableRecord.objectId(), "orders_segment"); + CatalogRecord original = record(consensus, history, MetadataProtocol.RecordKind.OFFLINE_SEGMENT_CONTENT, + contentName); + OfflineSegmentProtocol.Content content = OfflineSegmentProtocol.decodeContent(original.value()); + assertEquals(content.contentSha256(), HexFormat.of().formatHex( + MessageDigest.getInstance("SHA-256").digest(Files.readAllBytes(prepared.archive())))); + Path stored = Path.of(URI.create(content.downloadUri())); + assertTrue(stored.toRealPath().startsWith(directory.resolve("upload-storage").toRealPath())); + assertEquals(Files.readAllBytes(stored), Files.readAllBytes(prepared.archive())); + ReplicaInstallation installation = awaitInstallation(config, serverSession, content); + + response = upload(http, secondControllerUrl, "orders", prepared.archive(), true); + assertEquals(response.statusCode(), 200, response.body()); + PreparedSegment conflict = prepareSizeTestSegment(directory.resolve("conflicting-upload"), table, schema, 3L); + response = upload(http, controllerUrl, table.getTableName(), conflict.archive(), true); + assertEquals(response.statusCode(), 409, response.body()); + assertEquals(record(consensus, history, MetadataProtocol.RecordKind.OFFLINE_SEGMENT_CONTENT, contentName), + original, "Replay and conflicting bytes must leave the committed immutable content unchanged"); + assertEquals(awaitInstallation(config, serverSession, content), installation, + "Upload replay must reuse the existing assignment, operation, and installation receipt"); + return new InstalledOutput(content, prepared.bytes(), installation); + } + } + + private record PreparedSegment(Path archive, long bytes) { + } + + private static PreparedSegment prepareSizeTestSegment(Path directory, TableConfig table, Schema schema, + long secondId) throws Exception { Path generated = Files.createDirectories(directory.resolve("generated")); String segmentName = "orders_segment"; SegmentGeneratorConfig generator = new SegmentGeneratorConfig(table, schema); @@ -650,7 +712,7 @@ public class RatisStartersTest { GenericRow first = new GenericRow(); first.putValues(Map.of("id", 1L, "label", "one")); GenericRow second = new GenericRow(); - second.putValues(Map.of("id", 2L, "label", "two")); + second.putValues(Map.of("id", secondId, "label", "two")); driver.init(generator, new GenericRowRecordReader(List.of(first, second))); driver.build(); Path segment = generated.resolve(segmentName); @@ -658,71 +720,7 @@ public class RatisStartersTest { TarCompressionUtils.createCompressedTarFile(segment.toFile(), archive.toFile()); long expectedBytes = FileUtils.sizeOfDirectory(segment.resolve("v3").toFile()); assertTrue(expectedBytes > 0); - UUID history = UUID.fromString(serverSession.getHistoryId()); - UUID segmentId = UUID.randomUUID(); - try (RatisMetadataStore.Client consensus = RatisMetadataStore.connect(config)) { - MetadataProtocol.Result registered = MetadataProtocol.decodeResult(consensus.execute(MetadataProtocol.encode( - new MetadataProtocol.RegisterClient(history, "size-test", UUID.randomUUID(), 0)))); - assertEquals(registered.status(), MetadataProtocol.Status.OK, registered.detail()); - MetadataRpc.MembershipResponse controller = metadata.membershipExecute(MetadataRpc.MembershipCommandRequest - .newBuilder().setHistoryId(history.toString()).setRegister(MetadataRpc.ParticipantRegistration.newBuilder() - .setIncarnation(UUID.randomUUID().toString())).build()); - assertEquals(controller.getStatus(), ResultStatus.OK); - MetadataRpc.ParticipantToken session = controller.getParticipant().getParticipant(); - MetadataRpc.MembershipResponse ready = metadata.membershipExecute(MetadataRpc.MembershipCommandRequest - .newBuilder().setHistoryId(history.toString()).setReady(MetadataRpc.ParticipantReadiness.newBuilder() - .setSession(MetadataRpc.ParticipantSession.newBuilder().setIncarnation(session.getIncarnation()) - .setRegistrationEpoch(session.getRegistrationEpoch())) - .setAppliedRevision(controller.getRevision()).setReady(true)).build()); - assertEquals(ready.getStatus(), ResultStatus.OK); - MembershipProtocol.ControlToken control = MetadataGrpcClient.decodeMembershipResponse(ready) - .control().controlEpoch(); - OwnershipProtocol.Layout layout = new OwnershipProtocol.Layout(1, 24, - OwnershipProtocol.HashVersion.MURMUR2_UTF8_MASK_POSITIVE_V1); - commit(consensus, new MetadataProtocol.OwnershipCommand(history, "size-test", 1, "ownership", 1, - new OwnershipProtocol.ConfigureLayout(control, layout))); - OwnershipProtocol.TableShard shard = OwnershipProtocol.tableScope(layout, table.getTableName()); - MembershipProtocol.ParticipantToken controllerToken = MetadataGrpcClient.decodeMembershipResponse(ready) - .participant().participant(); - commit(consensus, new MetadataProtocol.OwnershipCommand(history, "size-test", 1, "ownership", 2, - new OwnershipProtocol.AssignOwner(control, 1, shard, 0, controllerToken))); - OwnershipProtocol.OwnerToken owner = new OwnershipProtocol.OwnerToken(history, 1, shard, 1, controllerToken); - CatalogRecord tableRecord = record(consensus, history, MetadataProtocol.RecordKind.TABLE_CONFIG, - table.getTableName()); - CatalogRecord schemaRecord = record(consensus, history, MetadataProtocol.RecordKind.SCHEMA, - schema.getSchemaName()); - ObjectVersion tableVersion = new ObjectVersion(history, tableRecord.objectId(), tableRecord.version()); - ObjectVersion schemaVersion = new ObjectVersion(history, schemaRecord.objectId(), schemaRecord.version()); - String digest = HexFormat.of().formatHex( - MessageDigest.getInstance("SHA-256").digest(Files.readAllBytes(archive))); - OfflineSegmentProtocol.Content content = new OfflineSegmentProtocol.Content(table.getTableName(), - tableRecord.objectId(), segmentName, segmentId, digest, archive.toUri().toString(), - new OfflineSegmentProtocol.LoaderDescriptor(new SegmentMetadataImpl(segment.toFile()).getCrc(), 0, false, - null, null, 100, 200, Map.of())); - MembershipProtocol.ParticipantToken server = new MembershipProtocol.ParticipantToken(history, - serverSession.getParticipantId(), Role.SERVER, UUID.fromString(serverSession.getIncarnation()), - serverSession.getRegistrationEpoch()); - commit(consensus, new MetadataProtocol.OfflineSegmentCommand(history, "size-test", 1, "segments", 1, - new OfflineSegmentProtocol.PublishContent(owner, tableVersion, content))); - commit(consensus, new MetadataProtocol.OfflineSegmentCommand(history, "size-test", 1, "segments", 2, - new OfflineSegmentProtocol.AssignReplica(owner, segmentId, 0, server, - new OfflineSegmentProtocol.Compatibility(tableVersion, schemaVersion), true))); - OfflineSegmentProtocol.Assignment assignment = OfflineSegmentProtocol.decodeAssignment(record(consensus, history, - MetadataProtocol.RecordKind.OFFLINE_REPLICA_ASSIGNMENT, - OfflineSegmentProtocol.assignmentName(segmentId, SERVER)).value()); - UUID operation = UUID.randomUUID(); - OperationProtocol.OfflineSegmentReload intent = new OperationProtocol.OfflineSegmentReload(table.getTableName(), - tableRecord.objectId(), segmentName, segmentId, digest, List.of(new OperationProtocol.TargetMaterialization( - server, assignment.assignmentEpoch(), assignment.materializationId())), true, true, false); - commit(consensus, new MetadataProtocol.NativeOperationCommand(history, "size-test", 1, "load", 1, - new OperationProtocol.Submit(operation, intent))); - OperationProtocol.OperationState accepted = OperationProtocol.decodeOperation(record(consensus, history, - MetadataProtocol.RecordKind.CONTROLLER_OPERATION, operation.toString()).value()); - commit(consensus, new MetadataProtocol.NativeOperationCommand(history, "size-test", 1, "load", 2, - new OperationProtocol.Claim(operation, 1, owner, - ((OperationProtocol.AssignedOfflineSegmentReload) accepted.input()).dependencies()))); - } - return expectedBytes; + return new PreparedSegment(archive, expectedBytes); } private static CatalogRecord record(RatisMetadataStore.Client client, UUID history, MetadataProtocol.RecordKind kind, @@ -733,15 +731,6 @@ public class RatisStartersTest { return result.record(); } - private static void commit(RatisMetadataStore.Client client, MetadataProtocol.SequencedCommand command) - throws IOException { - MetadataProtocol.Result result = MetadataProtocol.decodeResult(client.execute(MetadataProtocol.encode(command))); - assertEquals(result.status(), MetadataProtocol.Status.OK, result.detail()); - result = MetadataProtocol.decodeResult(client.execute(MetadataProtocol.encode(new MetadataProtocol.Acknowledge( - command.historyId(), command.clientId(), command.clientEpoch(), command.streamId(), command.sequence())))); - assertEquals(result.status(), MetadataProtocol.Status.OK, result.detail()); - } - private static void awaitSize(HttpClient client, String url, String field, long expectedBytes) throws Exception { long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(15); HttpResponse<String> response; @@ -819,6 +808,30 @@ public class RatisStartersTest { return client.send(request.build(), HttpResponse.BodyHandlers.ofString()); } + private static HttpResponse<String> upload(HttpClient client, String controllerUrl, String tableName, Path archive, + boolean authenticated) throws Exception { + return upload(client, controllerUrl, tableName, archive, authenticated, Map.of()); + } + + private static HttpResponse<String> upload(HttpClient client, String controllerUrl, String tableName, Path archive, + boolean authenticated, Map<String, String> headers) throws Exception { + String boundary = "native-upload-" + UUID.randomUUID(); + String header = "--" + boundary + "\r\n" + + "Content-Disposition: form-data; name=\"segment\"; filename=\"segment.tar.gz\"\r\n" + + "Content-Type: application/octet-stream\r\n\r\n"; + HttpRequest.BodyPublisher body = HttpRequest.BodyPublishers.concat(HttpRequest.BodyPublishers.ofString(header), + HttpRequest.BodyPublishers.ofFile(archive), + HttpRequest.BodyPublishers.ofString("\r\n--" + boundary + "--\r\n")); + HttpRequest.Builder request = HttpRequest.newBuilder(URI.create(controllerUrl + "/v2/segments?tableName=" + + tableName + "&tableType=OFFLINE&allowRefresh=false")).timeout(Duration.ofSeconds(15)) + .header("Content-Type", "multipart/form-data; boundary=" + boundary).POST(body); + if (authenticated) { + request.header("Authorization", authorization()); + } + headers.forEach(request::header); + return client.send(request.build(), HttpResponse.BodyHandlers.ofString()); + } + private static String authorization() { return "Basic " + Base64.getEncoder().encodeToString("admin:test-password".getBytes(StandardCharsets.UTF_8)); } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
