From aaf5a310c5ac999c81498308fdc11d6d5171463d Mon Sep 17 00:00:00 2001 From: sivabalan Date: Tue, 14 Nov 2023 19:37:34 -0800 Subject: [PATCH] Making perf enhancements to JSON serde --- .../TimelineServerBasedWriteMarkers.java | 10 ++-- .../apache/hudi/util/HttpRequestClient.java | 6 +- hudi-common/pom.xml | 6 ++ .../common/table/timeline/dto/DTOUtils.java | 4 +- .../view/RemoteHoodieTableFileSystemView.java | 56 +++++++++++-------- .../hudi/timeline/service/RequestHandler.java | 3 +- .../service/handlers/BaseFileHandler.java | 7 +-- .../handlers/marker/MarkerDirState.java | 3 +- pom.xml | 12 +++- 9 files changed, 66 insertions(+), 41 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/TimelineServerBasedWriteMarkers.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/TimelineServerBasedWriteMarkers.java index 2306763beb80d..85b213d303ddf 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/TimelineServerBasedWriteMarkers.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/TimelineServerBasedWriteMarkers.java @@ -62,6 +62,8 @@ */ public class TimelineServerBasedWriteMarkers extends WriteMarkers { private static final Logger LOG = LoggerFactory.getLogger(TimelineServerBasedWriteMarkers.class); + private static final TypeReference BOOLEAN_TYPE_REFERENCE = new TypeReference() {}; + private static final TypeReference> SET_TYPE_REFERENCE = new TypeReference>() {}; private final HttpRequestClient httpRequestClient; @@ -84,7 +86,7 @@ public boolean deleteMarkerDir(HoodieEngineContext context, int parallelism) { Map paramsMap = Collections.singletonMap(MARKER_DIR_PATH_PARAM, markerDirPath.toString()); try { return httpRequestClient.executeRequest( - DELETE_MARKER_DIR_URL, paramsMap, new TypeReference() {}, RequestMethod.POST); + DELETE_MARKER_DIR_URL, paramsMap, BOOLEAN_TYPE_REFERENCE, RequestMethod.POST); } catch (IOException e) { throw new HoodieRemoteException("Failed to delete marker directory " + markerDirPath.toString(), e); } @@ -95,7 +97,7 @@ public boolean doesMarkerDirExist() { Map paramsMap = Collections.singletonMap(MARKER_DIR_PATH_PARAM, markerDirPath.toString()); try { return httpRequestClient.executeRequest( - MARKERS_DIR_EXISTS_URL, paramsMap, new TypeReference() {}, RequestMethod.GET); + MARKERS_DIR_EXISTS_URL, paramsMap, BOOLEAN_TYPE_REFERENCE, RequestMethod.GET); } catch (IOException e) { throw new HoodieRemoteException("Failed to check marker directory " + markerDirPath.toString(), e); } @@ -106,7 +108,7 @@ public Set createdAndMergedDataPaths(HoodieEngineContext context, int pa Map paramsMap = Collections.singletonMap(MARKER_DIR_PATH_PARAM, markerDirPath.toString()); try { Set markerPaths = httpRequestClient.executeRequest( - CREATE_AND_MERGE_MARKERS_URL, paramsMap, new TypeReference>() {}, RequestMethod.GET); + CREATE_AND_MERGE_MARKERS_URL, paramsMap, SET_TYPE_REFERENCE, RequestMethod.GET); return markerPaths.stream().map(WriteMarkers::stripMarkerSuffix).collect(Collectors.toSet()); } catch (IOException e) { throw new HoodieRemoteException("Failed to get CREATE and MERGE data file paths in " @@ -119,7 +121,7 @@ public Set allMarkerFilePaths() { Map paramsMap = Collections.singletonMap(MARKER_DIR_PATH_PARAM, markerDirPath.toString()); try { return httpRequestClient.executeRequest( - ALL_MARKERS_URL, paramsMap, new TypeReference>() {}, RequestMethod.GET); + ALL_MARKERS_URL, paramsMap, SET_TYPE_REFERENCE, RequestMethod.GET); } catch (IOException e) { throw new HoodieRemoteException("Failed to get all markers in " + markerDirPath.toString(), e); } diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/util/HttpRequestClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/util/HttpRequestClient.java index 65131cc774283..6255f662c4ccd 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/util/HttpRequestClient.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/util/HttpRequestClient.java @@ -23,6 +23,7 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.module.afterburner.AfterburnerModule; import org.apache.http.client.fluent.Request; import org.apache.http.client.fluent.Response; import org.apache.http.client.utils.URIBuilder; @@ -37,7 +38,7 @@ */ public class HttpRequestClient { private static final Logger LOG = LoggerFactory.getLogger(HttpRequestClient.class); - private final ObjectMapper mapper; + private static final ObjectMapper MAPPER = new ObjectMapper().registerModule(new AfterburnerModule()); private final String serverHost; private final int serverPort; private final int timeoutSecs; @@ -51,7 +52,6 @@ public HttpRequestClient(HoodieWriteConfig writeConfig) { } public HttpRequestClient(String serverHost, int serverPort, int timeoutSecs, int maxRetry) { - this.mapper = new ObjectMapper(); this.serverHost = serverHost; this.serverPort = serverPort; this.timeoutSecs = timeoutSecs; @@ -92,7 +92,7 @@ public T executeRequest(String requestPath, Map queryParamet break; } String content = response.returnContent().asString(); - return (T) mapper.readValue(content, reference); + return (T) MAPPER.readValue(content, reference); } public enum RequestMethod { diff --git a/hudi-common/pom.xml b/hudi-common/pom.xml index b217d892eb61a..2474b007a241a 100644 --- a/hudi-common/pom.xml +++ b/hudi-common/pom.xml @@ -128,6 +128,12 @@ jackson-datatype-jsr310 + + + com.fasterxml.jackson.module + jackson-module-afterburner + + org.apache.avro diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/dto/DTOUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/dto/DTOUtils.java index ef5a886948765..4399860d6b4bb 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/dto/DTOUtils.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/dto/DTOUtils.java @@ -41,9 +41,9 @@ public static List fileGroupDTOsfromFileGroups(List fileGroupDTOS = new ArrayList<>(); + List fileGroupDTOS = new ArrayList<>(fileGroups.size()); fileGroupDTOS.add(FileGroupDTO.fromFileGroup(fileGroups.get(0), true)); - fileGroupDTOS.addAll(fileGroups.subList(1, fileGroups.size()).stream() + fileGroupDTOS.addAll(fileGroups.stream().skip(1) .map(fg -> FileGroupDTO.fromFileGroup(fg, false)).collect(Collectors.toList())); return fileGroupDTOS; } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java index f42f9bf2216cc..51fdeba9a37ce 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java @@ -43,6 +43,7 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.module.afterburner.AfterburnerModule; import org.apache.http.Consts; import org.apache.http.client.fluent.Request; import org.apache.http.client.fluent.Response; @@ -63,6 +64,8 @@ */ public class RemoteHoodieTableFileSystemView implements SyncableFileSystemView, Serializable { + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper().registerModule(new AfterburnerModule()); + private static final String BASE_URL = "/v1/hoodie/view"; public static final String LATEST_PARTITION_SLICES_URL = String.format("%s/%s", BASE_URL, "slices/partition/latest/"); public static final String LATEST_PARTITION_SLICE_URL = String.format("%s/%s", BASE_URL, "slices/file/latest/"); @@ -112,7 +115,6 @@ public class RemoteHoodieTableFileSystemView implements SyncableFileSystemView, public static final String PENDING_CLUSTERING_FILEGROUPS = String.format("%s/%s", BASE_URL, "clustering/pending/"); - public static final String LAST_INSTANT = String.format("%s/%s", BASE_URL, "timeline/instant/last"); public static final String LAST_INSTANTS = String.format("%s/%s", BASE_URL, "timeline/instants/last"); @@ -136,13 +138,20 @@ public class RemoteHoodieTableFileSystemView implements SyncableFileSystemView, private static final Logger LOG = LoggerFactory.getLogger(RemoteHoodieTableFileSystemView.class); + private static final TypeReference> FILE_SLICE_DTOS_REFERENCE = new TypeReference>() {}; + private static final TypeReference> FILE_GROUP_DTOS_REFERENCE = new TypeReference>() {}; + private static final TypeReference BOOLEAN_TYPE_REFERENCE = new TypeReference() {}; + private static final TypeReference> COMPACTION_OP_DTOS_REFERENCE = new TypeReference>() {}; + private static final TypeReference> CLUSTERING_OP_DTOS_REFERENCE = new TypeReference>() {}; + private static final TypeReference> INSTANT_DTOS_REFERENCE = new TypeReference>() {}; + private static final TypeReference TIMELINE_DTO_REFERENCE = new TypeReference() {}; + private static final TypeReference> BASE_FILE_DTOS_REFERENCE = new TypeReference>() {}; private final String serverHost; private final int serverPort; private final String basePath; private final HoodieTableMetaClient metaClient; private HoodieTimeline timeline; - private final ObjectMapper mapper; private final int timeoutMs; private boolean closed = false; @@ -159,7 +168,6 @@ public RemoteHoodieTableFileSystemView(String server, int port, HoodieTableMetaC public RemoteHoodieTableFileSystemView(HoodieTableMetaClient metaClient, FileSystemViewStorageConfig viewConf) { this.basePath = metaClient.getBasePath(); - this.mapper = new ObjectMapper(); this.metaClient = metaClient; this.timeline = metaClient.getActiveTimeline().filterCompletedAndCompactionInstants(); this.serverHost = viewConf.getRemoteViewServerHost(); @@ -192,7 +200,7 @@ private T executeRequest(String requestPath, Map queryParame LOG.info("Sending request : (" + url + ")"); Response response = retryHelper != null ? retryHelper.start(() -> get(timeoutMs, url, method)) : get(timeoutMs, url, method); String content = response.returnContent().asString(Consts.UTF_8); - return (T) mapper.readValue(content, reference); + return (T) OBJECT_MAPPER.readValue(content, reference); } private Map getParamsWithPartitionPath(String partitionPath) { @@ -250,7 +258,7 @@ public Stream getLatestBaseFiles() { private Stream getLatestBaseFilesFromParams(Map paramsMap, String requestPath) { try { List dataFiles = executeRequest(requestPath, paramsMap, - new TypeReference>() {}, RequestMethod.GET); + BASE_FILE_DTOS_REFERENCE, RequestMethod.GET); return dataFiles.stream().map(BaseFileDTO::toHoodieBaseFile); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -317,7 +325,7 @@ public Stream getLatestFileSlices(String partitionPath) { Map paramsMap = getParamsWithPartitionPath(partitionPath); try { List dataFiles = executeRequest(LATEST_PARTITION_SLICES_URL, paramsMap, - new TypeReference>() {}, RequestMethod.GET); + FILE_SLICE_DTOS_REFERENCE, RequestMethod.GET); return dataFiles.stream().map(FileSliceDTO::toFileSlice); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -329,7 +337,7 @@ public Option getLatestFileSlice(String partitionPath, String fileId) Map paramsMap = getParamsWithAdditionalParam(partitionPath, FILEID_PARAM, fileId); try { List dataFiles = executeRequest(LATEST_PARTITION_SLICE_URL, paramsMap, - new TypeReference>() {}, RequestMethod.GET); + FILE_SLICE_DTOS_REFERENCE, RequestMethod.GET); return Option.fromJavaOptional(dataFiles.stream().map(FileSliceDTO::toFileSlice).findFirst()); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -341,7 +349,7 @@ public Stream getLatestUnCompactedFileSlices(String partitionPath) { Map paramsMap = getParamsWithPartitionPath(partitionPath); try { List dataFiles = executeRequest(LATEST_PARTITION_UNCOMPACTED_SLICES_URL, paramsMap, - new TypeReference>() {}, RequestMethod.GET); + FILE_SLICE_DTOS_REFERENCE, RequestMethod.GET); return dataFiles.stream().map(FileSliceDTO::toFileSlice); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -356,7 +364,7 @@ public Stream getLatestFileSlicesBeforeOrOn(String partitionPath, Str new String[] {maxCommitTime, String.valueOf(includeFileSlicesInPendingCompaction)}); try { List dataFiles = executeRequest(LATEST_SLICES_BEFORE_ON_INSTANT_URL, paramsMap, - new TypeReference>() {}, RequestMethod.GET); + FILE_SLICE_DTOS_REFERENCE, RequestMethod.GET); return dataFiles.stream().map(FileSliceDTO::toFileSlice); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -386,7 +394,7 @@ public Stream getLatestMergedFileSlicesBeforeOrOn(String partitionPat Map paramsMap = getParamsWithAdditionalParam(partitionPath, MAX_INSTANT_PARAM, maxInstantTime); try { List dataFiles = executeRequest(LATEST_SLICES_MERGED_BEFORE_ON_INSTANT_URL, paramsMap, - new TypeReference>() {}, RequestMethod.GET); + FILE_SLICE_DTOS_REFERENCE, RequestMethod.GET); return dataFiles.stream().map(FileSliceDTO::toFileSlice); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -399,7 +407,7 @@ public Stream getLatestFileSliceInRange(List commitsToReturn) getParams(INSTANTS_PARAM, StringUtils.join(commitsToReturn.toArray(new String[0]), ",")); try { List dataFiles = executeRequest(LATEST_SLICES_RANGE_INSTANT_URL, paramsMap, - new TypeReference>() {}, RequestMethod.GET); + FILE_SLICE_DTOS_REFERENCE, RequestMethod.GET); return dataFiles.stream().map(FileSliceDTO::toFileSlice); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -411,7 +419,7 @@ public Stream getAllFileSlices(String partitionPath) { Map paramsMap = getParamsWithPartitionPath(partitionPath); try { List dataFiles = - executeRequest(ALL_SLICES_URL, paramsMap, new TypeReference>() {}, RequestMethod.GET); + executeRequest(ALL_SLICES_URL, paramsMap, FILE_SLICE_DTOS_REFERENCE, RequestMethod.GET); return dataFiles.stream().map(FileSliceDTO::toFileSlice); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -423,7 +431,7 @@ public Stream getAllFileGroups(String partitionPath) { Map paramsMap = getParamsWithPartitionPath(partitionPath); try { List fileGroups = executeRequest(ALL_FILEGROUPS_FOR_PARTITION_URL, paramsMap, - new TypeReference>() {}, RequestMethod.GET); + FILE_GROUP_DTOS_REFERENCE, RequestMethod.GET); return DTOUtils.fileGroupDTOsToFileGroups(fileGroups, metaClient); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -435,7 +443,7 @@ public Stream getReplacedFileGroupsBeforeOrOn(String maxCommitT Map paramsMap = getParamsWithAdditionalParam(partitionPath, MAX_INSTANT_PARAM, maxCommitTime); try { List fileGroups = executeRequest(ALL_REPLACED_FILEGROUPS_BEFORE_OR_ON, paramsMap, - new TypeReference>() {}, RequestMethod.GET); + FILE_GROUP_DTOS_REFERENCE, RequestMethod.GET); return DTOUtils.fileGroupDTOsToFileGroups(fileGroups, metaClient); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -447,7 +455,7 @@ public Stream getReplacedFileGroupsBefore(String maxCommitTime, Map paramsMap = getParamsWithAdditionalParam(partitionPath, MAX_INSTANT_PARAM, maxCommitTime); try { List fileGroups = executeRequest(ALL_REPLACED_FILEGROUPS_BEFORE, paramsMap, - new TypeReference>() {}, RequestMethod.GET); + FILE_GROUP_DTOS_REFERENCE, RequestMethod.GET); return DTOUtils.fileGroupDTOsToFileGroups(fileGroups, metaClient); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -459,7 +467,7 @@ public Stream getReplacedFileGroupsAfterOrOn(String minCommitTi Map paramsMap = getParamsWithAdditionalParam(partitionPath, MIN_INSTANT_PARAM, minCommitTime); try { List fileGroups = executeRequest(ALL_REPLACED_FILEGROUPS_AFTER_OR_ON, paramsMap, - new TypeReference>() {}, RequestMethod.GET); + FILE_GROUP_DTOS_REFERENCE, RequestMethod.GET); return DTOUtils.fileGroupDTOsToFileGroups(fileGroups, metaClient); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -471,7 +479,7 @@ public Stream getAllReplacedFileGroups(String partitionPath) { Map paramsMap = getParamsWithPartitionPath(partitionPath); try { List fileGroups = executeRequest(ALL_REPLACED_FILEGROUPS_PARTITION, paramsMap, - new TypeReference>() {}, RequestMethod.GET); + FILE_GROUP_DTOS_REFERENCE, RequestMethod.GET); return DTOUtils.fileGroupDTOsToFileGroups(fileGroups, metaClient); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -483,7 +491,7 @@ public boolean refresh() { try { // refresh the local timeline first. this.timeline = metaClient.reloadActiveTimeline().filterCompletedAndCompactionInstants(); - return executeRequest(REFRESH_TABLE, paramsMap, new TypeReference() {}, RequestMethod.POST); + return executeRequest(REFRESH_TABLE, paramsMap, BOOLEAN_TYPE_REFERENCE, RequestMethod.POST); } catch (IOException e) { throw new HoodieRemoteException(e); } @@ -493,7 +501,7 @@ public boolean refresh() { public Void loadAllPartitions() { Map paramsMap = getParams(); try { - executeRequest(LOAD_ALL_PARTITIONS_URL, paramsMap, new TypeReference() {}, RequestMethod.POST); + executeRequest(LOAD_ALL_PARTITIONS_URL, paramsMap, BOOLEAN_TYPE_REFERENCE, RequestMethod.POST); return null; } catch (IOException e) { throw new HoodieRemoteException(e); @@ -505,7 +513,7 @@ public Stream> getPendingCompactionOperations( Map paramsMap = getParams(); try { List dtos = executeRequest(PENDING_COMPACTION_OPS, paramsMap, - new TypeReference>() {}, RequestMethod.GET); + COMPACTION_OP_DTOS_REFERENCE, RequestMethod.GET); return dtos.stream().map(CompactionOpDTO::toCompactionOperation); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -517,7 +525,7 @@ public Stream> getPendingLogCompactionOperatio Map paramsMap = getParams(); try { List dtos = executeRequest(PENDING_LOG_COMPACTION_OPS, paramsMap, - new TypeReference>() {}, RequestMethod.GET); + COMPACTION_OP_DTOS_REFERENCE, RequestMethod.GET); return dtos.stream().map(CompactionOpDTO::toCompactionOperation); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -529,7 +537,7 @@ public Stream> getFileGroupsInPendingClus Map paramsMap = getParams(); try { List dtos = executeRequest(PENDING_CLUSTERING_FILEGROUPS, paramsMap, - new TypeReference>() {}, RequestMethod.GET); + CLUSTERING_OP_DTOS_REFERENCE, RequestMethod.GET); return dtos.stream().map(ClusteringOpDTO::toClusteringOperation); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -551,7 +559,7 @@ public Option getLastInstant() { Map paramsMap = getParams(); try { List instants = - executeRequest(LAST_INSTANT, paramsMap, new TypeReference>() {}, RequestMethod.GET); + executeRequest(LAST_INSTANT, paramsMap, INSTANT_DTOS_REFERENCE, RequestMethod.GET); return Option.fromJavaOptional(instants.stream().map(InstantDTO::toInstant).findFirst()); } catch (IOException e) { throw new HoodieRemoteException(e); @@ -563,7 +571,7 @@ public HoodieTimeline getTimeline() { Map paramsMap = getParams(); try { TimelineDTO timeline = - executeRequest(TIMELINE, paramsMap, new TypeReference() {}, RequestMethod.GET); + executeRequest(TIMELINE, paramsMap, TIMELINE_DTO_REFERENCE, RequestMethod.GET); return TimelineDTO.toTimeline(timeline, metaClient); } catch (IOException e) { throw new HoodieRemoteException(e); diff --git a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java index 345070fbe5c0e..4bb16bf6d03c9 100644 --- a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java +++ b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java @@ -45,6 +45,7 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.module.afterburner.AfterburnerModule; import io.javalin.Javalin; import io.javalin.http.BadRequestResponse; import io.javalin.http.Context; @@ -69,7 +70,7 @@ */ public class RequestHandler { - private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper().registerModule(new AfterburnerModule()); private static final Logger LOG = LoggerFactory.getLogger(RequestHandler.class); private final TimelineService.Config timelineServiceConfig; diff --git a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/BaseFileHandler.java b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/BaseFileHandler.java index a34b49843fac1..2d0d4e71d5aa1 100644 --- a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/BaseFileHandler.java +++ b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/BaseFileHandler.java @@ -28,6 +28,7 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.stream.Collectors; @@ -74,10 +75,8 @@ public Map> getAllLatestDataFilesBeforeOrOn(String bas public List getLatestDataFileOn(String basePath, String partitionPath, String instantTime, String fileId) { - List result = new ArrayList<>(); - viewManager.getFileSystemView(basePath).getBaseFileOn(partitionPath, instantTime, fileId) - .map(BaseFileDTO::fromHoodieBaseFile).ifPresent(result::add); - return result; + return viewManager.getFileSystemView(basePath).getBaseFileOn(partitionPath, instantTime, fileId) + .map(BaseFileDTO::fromHoodieBaseFile).map(Collections::singletonList).orElse(Collections.emptyList()); } public List getLatestDataFilesInRange(String basePath, List instants) { diff --git a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java index 9f8ed5d84cfe9..05551dc42dde3 100644 --- a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java +++ b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java @@ -32,6 +32,7 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.module.afterburner.AfterburnerModule; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; @@ -64,7 +65,7 @@ */ public class MarkerDirState implements Serializable { private static final Logger LOG = LoggerFactory.getLogger(MarkerDirState.class); - private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper().registerModule(new AfterburnerModule()); // Marker directory private final String markerDirPath; private final FileSystem fileSystem; diff --git a/pom.xml b/pom.xml index fe49188a2f82a..8ac7f7f65dca4 100644 --- a/pom.xml +++ b/pom.xml @@ -471,6 +471,8 @@ org.apache.hbase.thirdparty:hbase-shaded-netty org.apache.hbase.thirdparty:hbase-shaded-protobuf org.apache.htrace:htrace-core4 + + com.fasterxml.jackson.module:jackson-module-afterburner @@ -873,6 +875,12 @@ ${fasterxml.jackson.module.scala.version} + + com.fasterxml.jackson.module + jackson-module-afterburner + ${fasterxml.jackson.databind.version} + + org.glassfish.jersey.core @@ -2174,7 +2182,7 @@ 1.8.2 4.7 2.6.7 - 2.6.7.3 + ${fasterxml.version} 2.6.7.1 2.7.4 true @@ -2206,7 +2214,7 @@ 1.8.2 4.7 2.6.7 - 2.6.7.3 + ${fasterxml.version} 2.6.7.1 2.7.4 true