feat(cli): add the feature to turn of WRAP for CLI output

ksql-cli/src/main/java/io/confluent/ksql/cli/Cli.java

@@ -215,7 +215,9 @@ void handleLine(final String line) throws Exception {
       return;
     }
 
-    handleStatements(line);
+    if (!terminal.maybeHandleCliSpecificCommands(trimmedLine)) {
+      handleStatements(trimmedLine);
+    }
   }
 
   /**

ksql-cli/src/main/java/io/confluent/ksql/cli/console/CliConfig.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2019 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.cli.console;
+
+import io.confluent.ksql.configdef.ConfigValidators;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
+
+public class CliConfig extends AbstractConfig {
+
+  public static final String WRAP_CONFIG = "WRAP";
+
+  private static final ConfigDef CONFIG_DEF = new ConfigDef()
+      .define(
+          WRAP_CONFIG,
+          Type.STRING,
+          OnOff.ON.name(),
+          ConfigValidators.enumValues(OnOff.class),
+          Importance.MEDIUM,
+          "A value of 'OFF' will clip lines to ensure that query results do not exceed the "
+              + "terminal width (i.e. each row will appear on a single line)."
+      );
+
+  public CliConfig(final Map<?, ?> originals) {
+    super(CONFIG_DEF, originals);
+  }
+
+  public CliConfig with(final String property, final Object value) {
+    final Map<String, Object> originals = new HashMap<>(originals());
+    originals.put(property, value);
+    return new CliConfig(originals);
+  }
+
+  @SuppressWarnings("unused") // used in validation
+  public enum OnOff {
+    ON, OFF
+  }
+
+}

ksql-cli/src/main/java/io/confluent/ksql/cli/console/Console.java

@@ -19,6 +19,7 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import io.confluent.ksql.GenericRow;
 import io.confluent.ksql.cli.console.KsqlTerminal.HistoryEntry;
@@ -185,7 +186,7 @@ private static <T extends KsqlEntity> Handler1<KsqlEntity, Console> tablePrinter
   private final RowCaptor rowCaptor;
   private OutputFormat outputFormat;
   private Optional<File> spoolFile = Optional.empty();
-
+  private CliConfig config;
 
   public interface RowCaptor {
     void addRow(GenericRow row);
@@ -225,6 +226,7 @@ public Console(
     this.rowCaptor = Objects.requireNonNull(rowCaptor, "rowCaptor");
     this.cliSpecificCommands = Maps.newLinkedHashMap();
     this.objectMapper = JsonMapper.INSTANCE.mapper;
+    this.config = new CliConfig(ImmutableMap.of());
   }
 
   public PrintWriter writer() {
@@ -268,6 +270,10 @@ public void handle(final Signal signal, final SignalHandler signalHandler) {
     terminal.handle(signal, signalHandler);
   }
 
+  public void setCliProperty(final String name, final Object value) {
+    config = config.with(name, value);
+  }
+
   @Override
   public void close() {
     terminal.close();
@@ -282,14 +288,7 @@ public Map<String, CliSpecificCommand> getCliSpecificCommands() {
   }
 
   public String readLine() {
-    String line;
-
-    do {
-      line = terminal.readLine();
-
-    } while (maybeHandleCliSpecificCommands(line));
-
-    return line;
+    return terminal.readLine();
   }
 
   public List<HistoryEntry> getHistory() {
@@ -402,14 +401,11 @@ private Optional<CliCmdExecutor> getCliCommand(final String line) {
       return Optional.empty();
     }
 
-    final String reconstructed = parts.stream()
-        .collect(Collectors.joining(" "));
-
-    final String asLowerCase = reconstructed.toLowerCase();
+    final String command = String.join(" ", parts);
 
-    return cliSpecificCommands.entrySet().stream()
-        .filter(e -> asLowerCase.startsWith(e.getKey()))
-        .map(e -> CliCmdExecutor.of(e.getValue(), parts))
+    return cliSpecificCommands.values().stream()
+        .filter(cliSpecificCommand -> cliSpecificCommand.matches(command))
+        .map(cliSpecificCommand -> CliCmdExecutor.of(cliSpecificCommand, parts))
         .findFirst();
   }
 
@@ -418,7 +414,7 @@ private void printAsTable(
       final List<FieldInfo> fields
   ) {
     rowCaptor.addRow(row);
-    writer().println(TabularRow.createRow(getWidth(), fields, row));
+    writer().println(TabularRow.createRow(getWidth(), fields, row, config));
     flush();
   }
 
@@ -787,7 +783,7 @@ public void addRows(final List<List<String>> fields) {
     }
   }
 
-  private boolean maybeHandleCliSpecificCommands(final String line) {
+  public boolean maybeHandleCliSpecificCommands(final String line) {
     if (line == null) {
       return false;
     }

ksql-cli/src/main/java/io/confluent/ksql/cli/console/cmd/CliCommandRegisterUtil.java

@@ -68,5 +68,9 @@ public static void registerDefaultCommands(
 
     console.registerCliSpecificCommand(
         Spool.create(console::setSpool, console::unsetSpool));
+
+    console.registerCliSpecificCommand(
+        SetCliProperty.create(console::setCliProperty)
+    );
   }
 }

ksql-cli/src/main/java/io/confluent/ksql/cli/console/cmd/CliSpecificCommand.java

@@ -20,6 +20,14 @@
 
 public interface CliSpecificCommand {
 
+  /**
+   * @param command the full command
+   * @return whether or not {@code command} is an instance of {@code this}
+   */
+  default boolean matches(final String command) {
+    return command.toLowerCase().startsWith(getName().toLowerCase());
+  }
+
   /**
    * Get the name of the command.
    *

ksql-cli/src/main/java/io/confluent/ksql/cli/console/cmd/SetCliProperty.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2019 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.cli.console.cmd;
+
+import java.io.PrintWriter;
+import java.util.List;
+import java.util.function.BiConsumer;
+
+final class SetCliProperty implements CliSpecificCommand {
+
+  private static final String HELP = "set <property> <value>:" + System.lineSeparator()
+      + "\tSets a CLI local property. NOTE that this differs from setting a KSQL "
+      + "property with 'SET property=value' in that it does not affect the server.";
+
+  private final BiConsumer<String, String> setProperty;
+
+  public static SetCliProperty create(final BiConsumer<String, String> setProperty) {
+    return new SetCliProperty(setProperty);
+  }
+
+  private SetCliProperty(final BiConsumer<String, String> setProperty) {
+    this.setProperty = setProperty;
+  }
+
+  @Override
+  public boolean matches(final String command) {
+    return command.toLowerCase().startsWith(getName().toLowerCase()) && !command.contains("=");
+  }
+
+  @Override
+  public String getName() {
+    return "set";
+  }
+
+  @Override
+  public String getHelpMessage() {
+    return HELP;
+  }
+
+  @Override
+  public void execute(final List<String> args, final PrintWriter terminal) {
+    CliCmdUtil.ensureArgCountBounds(args, 2, 2, getHelpMessage());
+    setProperty.accept(args.get(0), args.get(1));
+  }
+}

ksql-cli/src/main/java/io/confluent/ksql/util/TabularRow.java

@@ -19,6 +19,8 @@
 import com.google.common.base.Splitter;
 import com.google.common.base.Strings;
 import io.confluent.ksql.GenericRow;
+import io.confluent.ksql.cli.console.CliConfig;
+import io.confluent.ksql.cli.console.CliConfig.OnOff;
 import io.confluent.ksql.rest.entity.FieldInfo;
 import java.util.ArrayList;
 import java.util.List;
@@ -27,42 +29,49 @@
 
 public class TabularRow {
 
+  private static final String CLIPPED = "...";
   private static final int MIN_CELL_WIDTH = 5;
 
   private final int width;
   private final List<String> value;
   private final List<String> header;
   private final boolean isHeader;
+  private final boolean shouldWrap;
 
   public static TabularRow createHeader(final int width, final List<FieldInfo> header) {
     return new TabularRow(
         width,
         header.stream().map(FieldInfo::getName).collect(Collectors.toList()),
-        null);
+        null,
+        true);
   }
 
   public static TabularRow createRow(
       final int width,
       final List<FieldInfo> header,
-      final GenericRow value
+      final GenericRow value,
+      final CliConfig config
   ) {
     return new TabularRow(
         width,
         header.stream().map(FieldInfo::getName).collect(Collectors.toList()),
-        value.getColumns().stream().map(Objects::toString).collect(Collectors.toList())
+        value.getColumns().stream().map(Objects::toString).collect(Collectors.toList()),
+        config.getString(CliConfig.WRAP_CONFIG).equalsIgnoreCase(OnOff.ON.toString())
     );
   }
 
   @VisibleForTesting
   TabularRow(
       final int width,
       final List<String> header,
-      final List<String> value
+      final List<String> value,
+      final boolean shouldWrap
   ) {
     this.header = Objects.requireNonNull(header, "header");
     this.width = width;
     this.value = value;
     this.isHeader = value == null;
+    this.shouldWrap = shouldWrap;
   }
 
   @Override
@@ -92,7 +101,7 @@ public String toString() {
         .map(s -> addUntil(s, createCell("", cellWidth), maxSplit))
         .collect(Collectors.toList());
 
-    formatRow(builder, buffered, maxSplit);
+    formatRow(builder, buffered, shouldWrap ? maxSplit : 1);
 
     if (isHeader) {
       builder.append('\n');
@@ -111,14 +120,31 @@ private static void formatRow(
     for (int row = 0; row < numRows; row++) {
       builder.append('|');
       for (int col = 0; col < columns.size(); col++) {
-        builder.append(columns.get(col).get(row));
+        final String colValue = columns.get(col).get(row);
+        if (shouldClip(columns.get(col), numRows)) {
+          builder.append(colValue, 0, colValue.length() - CLIPPED.length())
+              .append(CLIPPED)
+              .append('|');
+        } else {
+          builder.append(colValue)
+              .append('|');
+        }
       }
       if (row != numRows - 1) {
         builder.append('\n');
       }
     }
   }
 
+  private static boolean shouldClip(final List<String> parts, final int rowsToPrint) {
+    // clip if there are more than one line and any of the remaining lines are non-empty
+    return parts.size() > rowsToPrint
+        && parts.subList(rowsToPrint, parts.size())
+        .stream()
+        .map(String::trim)
+        .noneMatch(String::isEmpty);
+  }
+
   @SuppressWarnings("UnstableApiUsage")
   private static List<String> splitToFixed(final String value, final int width) {
     return Splitter.fixedLength(width)
@@ -141,7 +167,7 @@ private static void separatingLine(
   }
 
   private static String createCell(final String value, final int width) {
-    final String format = "%-" + width + "s|";
+    final String format = "%-" + width + "s";
     return String.format(format, value);
   }
 

ksql-cli/src/test/java/io/confluent/ksql/cli/CliTest.java

@@ -996,7 +996,7 @@ private void givenRequestPipelining(final String setting) {
 
   private void runCliSpecificCommand(final String command) {
     when(lineSupplier.get()).thenReturn(command).thenReturn("");
-    console.readLine();
+    console.maybeHandleCliSpecificCommands(console.readLine());
   }
 
   private void givenRunInteractivelyWillExit() {