Pipe: Set WAL to uncompressed when using real-time sync (#15733)

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/IoTDBDataRegionExtractor.java

@@ -53,6 +53,7 @@
 import org.apache.iotdb.pipe.api.exception.PipeException;
 import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException;
 
+import org.apache.tsfile.file.metadata.enums.CompressionType;
 import org.apache.tsfile.utils.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -468,7 +469,7 @@ private void constructRealtimeExtractor(final PipeParameters parameters)
     // Use hybrid mode by default
     if (!parameters.hasAnyAttributes(EXTRACTOR_MODE_STREAMING_KEY, SOURCE_MODE_STREAMING_KEY)
         && !parameters.hasAnyAttributes(EXTRACTOR_REALTIME_MODE_KEY, SOURCE_REALTIME_MODE_KEY)) {
-      checkWalEnable(parameters);
+      checkWalEnableAndSetUncompressed(parameters);
       realtimeExtractor = new PipeRealtimeDataRegionHybridExtractor();
       LOGGER.info(
           "Pipe: '{}' ('{}') and '{}' ('{}') is not set, use hybrid mode by default.",
@@ -485,7 +486,7 @@ private void constructRealtimeExtractor(final PipeParameters parameters)
               Arrays.asList(EXTRACTOR_MODE_STREAMING_KEY, SOURCE_MODE_STREAMING_KEY),
               EXTRACTOR_MODE_STREAMING_DEFAULT_VALUE);
       if (isStreamingMode) {
-        checkWalEnable(parameters);
+        checkWalEnableAndSetUncompressed(parameters);
         realtimeExtractor = new PipeRealtimeDataRegionHybridExtractor();
       } else {
         realtimeExtractor = new PipeRealtimeDataRegionTsFileExtractor();
@@ -501,15 +502,15 @@ private void constructRealtimeExtractor(final PipeParameters parameters)
       case EXTRACTOR_REALTIME_MODE_HYBRID_VALUE:
       case EXTRACTOR_REALTIME_MODE_LOG_VALUE:
       case EXTRACTOR_REALTIME_MODE_STREAM_MODE_VALUE:
-        checkWalEnable(parameters);
+        checkWalEnableAndSetUncompressed(parameters);
         realtimeExtractor = new PipeRealtimeDataRegionHybridExtractor();
         break;
       case EXTRACTOR_REALTIME_MODE_FORCED_LOG_VALUE:
-        checkWalEnable(parameters);
+        checkWalEnableAndSetUncompressed(parameters);
         realtimeExtractor = new PipeRealtimeDataRegionLogExtractor();
         break;
       default:
-        checkWalEnable(parameters);
+        checkWalEnableAndSetUncompressed(parameters);
         realtimeExtractor = new PipeRealtimeDataRegionHybridExtractor();
         if (LOGGER.isWarnEnabled()) {
           LOGGER.warn(
@@ -519,14 +520,26 @@ private void constructRealtimeExtractor(final PipeParameters parameters)
     }
   }
 
-  private void checkWalEnable(final PipeParameters parameters) throws IllegalPathException {
+  private void checkWalEnableAndSetUncompressed(final PipeParameters parameters)
+      throws IllegalPathException {
     if (Boolean.TRUE.equals(
             DataRegionListeningFilter.parseInsertionDeletionListeningOptionPair(parameters)
                 .getLeft())
         && IoTDBDescriptor.getInstance().getConfig().getWalMode().equals(WALMode.DISABLE)) {
       throw new PipeException(
           "The pipe cannot transfer realtime insertion if data region disables wal. Please set 'realtime.mode'='batch' in source parameters when enabling realtime transmission.");
     }
+
+    if (!IoTDBDescriptor.getInstance()
+        .getConfig()
+        .getWALCompressionAlgorithm()
+        .equals(CompressionType.UNCOMPRESSED)) {
+      LOGGER.info(
+          "The pipe prefers uncompressed wal, and may introduce certain delay in realtime insert syncing without it. Hence, we change it to uncompressed.");
+      IoTDBDescriptor.getInstance()
+          .getConfig()
+          .setWALCompressionAlgorithm(CompressionType.UNCOMPRESSED);
+    }
   }
 
   @Override